-
Notifications
You must be signed in to change notification settings - Fork 4
/
connections.py
80 lines (56 loc) · 3.15 KB
/
connections.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
#!/usr/bin/env python
# coding: utf-8
# In[ ]:
#DB Connections
import mysql.connector as sql_con
from sqlalchemy import create_engine
import cred_config as cc
import ftplib
from settings import server_type
from sqlalchemy.sql import text
# In[ ]:
def connections(con_name):
if(con_name=='prod_db_read'):
if(server_type=='local'):
cred_r=cc.credentials['covidsos.org-local']
server_con = sql_con.connect(user=cred_r['user'], password=cred_r['password'], host=cred_r['host'],database=cred_r['database'])
elif(server_type=='prod'):
cred_r=cc.credentials['covidsos.org-prod']
server_con = sql_con.connect(user=cred_r['user'], password=cred_r['password'], host=cred_r['host'],database=cred_r['database'])
elif(server_type=='staging'):
cred_r=cc.credentials['covidsos.org-staging']
server_con = sql_con.connect(user=cred_r['user'], password=cred_r['password'], host=cred_r['host'],database=cred_r['database'])
if(con_name=='prod_db_write'):
if(server_type=='local'):
cred_w=cc.credentials['covidsos.org-local']
server_con = create_engine("mysql+pymysql://{user}:{password}@{host}/{database}".format(user = cred_w['user'], password = cred_w['password'], host = cred_w['host'], database = cred_w['database']), pool_size=10, max_overflow=20, echo=False)
elif(server_type=='prod'):
cred_w=cc.credentials['covidsos.org-prod']
server_con = create_engine("mysql+pymysql://{user}:{password}@{host}/{database}".format(user = cred_w['user'], password = cred_w['password'], host = cred_w['host'], database = cred_w['database']), pool_size=10, max_overflow=20, echo=False)
elif(server_type=='staging'):
cred_w=cc.credentials['covidsos.org-staging']
server_con = create_engine("mysql+pymysql://{user}:{password}@{host}/{database}".format(user = cred_w['user'], password = cred_w['password'], host = cred_w['host'], database = cred_w['database']), pool_size=10, max_overflow=20, echo=False)
if(con_name=='db_read'):
cred_r=cc.credentials['covid_sos_read']
server_con = sql_con.connect(user=cred_r['user'], password=cred_r['password'], host=cred_r['host'],database=cred_r['database'])
if(con_name=='db_write'):
cred_w = cc.credentials['covid_sos_write']
server_con = create_engine("mysql+pymysql://{user}:{password}@{host}/{database}".format(user = cred_w['user'], password = cred_w['password'], host = cred_w['host'], database = cred_w['database']), pool_size=10, max_overflow=20, echo=False)
if(con_name=='ftp'):
FTP_con = cc.credentials['ftp']
server_con = ftplib.FTP(host=FTP_con['host'], user=FTP_con['user'], passwd=FTP_con['password'])
return server_con
def keys(api_name):
if(api_name=='gmap'):
gmap_key = cc.credentials['gmap']['key']
return gmap_key
# In[ ]:
def write_query(q,server_name):
if((server_name=='prod_db_write') or (server_name=='db_write')):
engine = connections(server_name)
with engine.connect() as con:
query = text(q)
con.execute(query)
return None
# In[ ]:
# In[ ]: