-
Notifications
You must be signed in to change notification settings - Fork 0
/
dbutils.py
53 lines (44 loc) · 2.04 KB
/
dbutils.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
from sqlalchemy import event, exc
from sqlalchemy.engine import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, session
from sqlalchemy.orm.session import sessionmaker
from sqlalchemy.pool import Pool
from sqlalchemy.schema import MetaData
from sqlalchemy.sql.expression import ClauseElement
#logger = logging.getLogger('dbutils')
def get_or_create(session, model, **kwargs):
instance = session.query(model).filter_by(**kwargs).first()
if instance:
return instance
else:
params = dict((k, v) for k, v in kwargs.iteritems() if not isinstance(v, ClauseElement))
instance = model(**params)
session.add(instance)
session.flush()
session.commit()
return instance
@event.listens_for(Pool, "checkout")
def ping_connection(dbapi_connection, connection_record, connection_proxy):
cursor = dbapi_connection.cursor()
try:
cursor.execute("SELECT 1")
except:
# optional - dispose the whole pool
# instead of invalidating one at a time
# connection_proxy._pool.dispose()
# raise DisconnectionError - pool will try
# connecting again up to three times before raising.
# logger.error('Connection died, reconnect attempt')
raise exc.DisconnectionError()
cursor.close()
#engine = create_engine('mysql://bfc1ffabdb36c3:65da212b@us-cdbr-east-02.cleardb.com/heroku_1cec684f35035ce?charset=utf8', echo=True, pool_recycle=3600)#recycle connection every hour to prevent overnight disconnect)
#Base = declarative_base(bind=engine)
#Base.metadata.create_all(engine)
#sm = sessionmaker(bind=engine, autoflush=True, autocommit=False, expire_on_commit=False)
#Session = scoped_session(sm)
#engine = create_engine('sqlite:///db3.sqlite', echo=True)
#logging.getLogger('sqlalchemy.engine').setLevel(logging.WARN)
#logging.getLogger('sqlalchemy.dialects').setLevel(logging.WARN)
#logging.getLogger('sqlalchemy.pool').setLevel(logging.WARN)
#logging.getLogger('sqlalchemy.orm').setLevel(logging.WARN)