openmedialibrary/oml/db.py

76 lines
1.9 KiB
Python
Raw Normal View History

2014-08-09 16:14:14 +00:00
from contextlib import contextmanager
from sqlalchemy import create_engine, MetaData
from sqlalchemy import orm
from sqlalchemy.orm.exc import UnmappedClassError
from sqlalchemy.orm import sessionmaker, scoped_session
2014-05-04 17:26:43 +00:00
from sqlalchemy.ext.mutable import Mutable
2014-08-09 15:03:16 +00:00
from sqlalchemy.ext.declarative import declarative_base
2014-08-09 16:14:14 +00:00
import settings
import state
2014-08-09 15:03:16 +00:00
engine = create_engine('sqlite:////%s' % settings.db_path)
2014-08-09 16:14:14 +00:00
Session = scoped_session(sessionmaker(bind=engine))
metadata = MetaData()
class _QueryProperty(object):
def __init__(self):
pass
2014-08-09 15:03:16 +00:00
2014-08-09 16:14:14 +00:00
def __get__(self, obj, type):
try:
mapper = orm.class_mapper(type)
if mapper:
return type.query_class(mapper, session=state.db.session)
except UnmappedClassError:
return None
2014-08-09 15:03:16 +00:00
2014-08-09 16:14:14 +00:00
class BaseQuery(orm.Query):
pass
2014-08-09 15:03:16 +00:00
Model = declarative_base()
2014-08-09 16:14:14 +00:00
Model.query_class = BaseQuery
Model.query = _QueryProperty()
Model.metadata = metadata
2014-08-09 15:03:16 +00:00
2014-08-09 16:14:14 +00:00
@contextmanager
def session():
if hasattr(state.db, 'session'):
state.db.count += 1
else:
state.db.session = Session()
state.db.count = 1
yield
state.db.count -= 1
if not state.db.count:
Session.remove()
2014-05-04 17:26:43 +00:00
class MutableDict(Mutable, dict):
@classmethod
def coerce(cls, key, value):
"Convert plain dictionaries to MutableDict."
if not isinstance(value, MutableDict):
if isinstance(value, dict):
return MutableDict(value)
# this call will raise ValueError
return Mutable.coerce(key, value)
else:
return value
def __setitem__(self, key, value):
"Detect dictionary set events and emit change events."
dict.__setitem__(self, key, value)
self.changed()
def __delitem__(self, key):
"Detect dictionary del events and emit change events."
dict.__delitem__(self, key)
self.changed()