-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathplsql.py
More file actions
83 lines (69 loc) · 2.18 KB
/
Copy pathplsql.py
File metadata and controls
83 lines (69 loc) · 2.18 KB
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
77
78
79
80
81
82
83
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy_utils import database_exists, create_database
import os,sys
dir_path = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.abspath(os.path.join(dir_path, os.pardir)))
from database.config import db_settings as settings
import logging
log = logging.getLogger(__name__)
def get_database():
"""
Connects to database.
Returns:
engine
"""
try:
engine = get_engine_from_settings()
log.info("Connected to PostgreSQL database!")
except IOError:
log.exception("Failed to get database connection!")
return None, 'fail'
return engine
def get_engine_from_settings():
"""
Sets up database connection from local settings.
Input:
Dictionary containing pghost, pguser, pgpassword, pgdatabase and pgport.
Returns:
Call to get_database returning engine
"""
keys = ['user','password','host','port','db']
if not all(key in keys for key in settings.keys()):
raise Exception('Bad config file')
return get_engine(settings['user'],
settings['password'],
settings['host'],
settings['port'],
settings['db'])
def get_engine(user, passwd, host, port, db):
"""
Get SQLalchemy engine using credentials.
Input:
db: database name
user: Username
host: Hostname of the database server
port: Port number
passwd: Password for the database
Returns:
Database engine
"""
url = 'postgresql://{user}:{passwd}@{host}:{port}/{db}'.format(
user=user, passwd=passwd, host=host, port=port, db=db)
if not database_exists(url):
create_database(url)
engine = create_engine(url, pool_size=50, echo=False)
return engine
def get_session():
"""
Return an SQLAlchemy session
Input:
engine: an SQLAlchemy engine
"""
engine = get_database()
session = sessionmaker(bind=engine)()
return session
db = get_database()
session = get_session()
Base = declarative_base()