'''
Base for interfacing with SQLAlchemy
Provides the necessary plumbing for CRUD
using SA session
'''
from inspector import class_keys, primary_key, KeyNotFoundException
[docs]class AlchemyModelManager(object):
def __init__(self, *args, **kwargs):
'''
self.session is expected to be initialized
by the derived class
'''
self.cls = self.model_class
assert hasattr(self, 'session'), "session is expected"
assert self.session is not None, "session must be initialized" \
"by the derived class"
[docs] def list(self, other_pks=None, filters=None):
'''
List returns back list of URI
In case of multiple pks, We guess the pk
by using '<modelname>_id' as the field convention
'''
try:
pk = primary_key(self.cls)
except KeyNotFoundException:
return list()
filter_dict = dict()
if filters:
filter_dict = {k: v for k, v in filters.iteritems()}
filter_dict.pop('format', None)
filter_dict.pop('page', None)
filter_dict.pop('sort_by', None)
if other_pks:
query_pks = dict()
# Use only those keys which are primary keys
# and pick them up from the passed dict
for key in class_keys(self.cls):
if key in other_pks:
query_pks[key] = other_pks[key]
query_pks.update(filter_dict)
queryset = self.session.query(self.cls.__dict__[pk]).filter_by(
**query_pks).all()
else:
if filter_dict:
queryset = self.session.query(self.cls.__dict__[pk]).filter_by(
**filter_dict).all()
else:
queryset = self.session.query(self.cls.__dict__[pk]).all()
return queryset
[docs] def retrieve(self, pks, other_pks=None):
'''
Retrieve fetches the object based on the following pk logic:
if 'other' pks are not found, just use the pk list (coming from URLS)
assuming their order is already correct
if 'other' pks are found, then use the class keys to
get the correct order of pks, look them up
'''
if not other_pks:
newargs = list(pks)
else:
newargs = list()
for key in class_keys(self.cls):
if other_pks and key in other_pks:
newargs.append(other_pks[key])
# Confirm this logic works!!!
# will the order be correct if we just append?
for pk in reversed(pks):
newargs.append(pk)
return self.session.query(self.cls).get(newargs)