marcus/PyJeeves
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Stored procedure helpers. Order repo and model. More SQLService updates.

Browse source 
* A generic stored procedure helper added, and support for calling them.
* Order and OrderItem tables added, including helpers and calls to SP for creation and updates.
* Minor updates to other repositories.
This commit is contained in:
Marcus Lindvall 2019-08-30 12:09:10 +02:00
parent b77a7069ce
commit 0af38e286e
9 changed files with 730 additions and 138 deletions
Download patch file Download diff file Expand all files Collapse all files

142
pyjeeves/connector.py
View file

@ -6,29 +6,31 @@
Global objects
"""
from pyjeeves import logging, config
from weakref import WeakValueDictionary
from sqlalchemy import create_engine, orm
from sqlalchemy.orm import sessionmaker, scoped_session, Query, aliased
from sqlalchemy.orm.exc import UnmappedClassError
from sqlalchemy.ext.declarative import declarative_base, DeclarativeMeta
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
# from sqlalchemy.orm.exc import UnmappedClassError
from pymssql import OperationalError
from sqlservice import SQLClient, SQLQuery
logger = logging.getLogger("PyJeeves." + __name__)
class BaseFilterQuery(Query):
class BaseFilterQuery(SQLQuery):
def get(self, ident):
# Override get() so that the flag is always checked in the
# DB as opposed to pulling from the identity map. - this is optional.
return Query.get(self.populate_existing(), ident)
return SQLQuery.get(self.populate_existing(), ident)
def __iter__(self):
return Query.__iter__(self.private())
return SQLQuery.__iter__(self.private())
def from_self(self, *ent):
# Override from_self() to automatically apply
# the criterion to. this works with count() and
# others.
return Query.from_self(self.private(), *ent)
return SQLQuery.from_self(self.private(), *ent)
def private(self):
# Fetch the model name and column list and apply model-specific base filters
@ -49,94 +51,58 @@ class BaseFilterQuery(Query):
return self
class Model(object):
"""Baseclass for custom user models."""
#: the query class used. The :attr:`query` attribute is an instance
#: of this class. By default a :class:`BaseQuery` is used.
query_class = BaseFilterQuery
#: an instance of :attr:`query_class`. Can be used to query the
#: database for instances of this model.
query = None
class MetaBaseModel(DeclarativeMeta):
""" Define a metaclass for the BaseModel
Implement `__getitem__` for managing aliases """
def __init__(cls, *args):
super().__init__(*args)
cls.aliases = WeakValueDictionary()
def __getitem__(cls, key):
try:
alias = cls.aliases[key]
except KeyError:
alias = aliased(cls)
cls.aliases[key] = alias
return alias
class _QueryProperty(object):
def __init__(self, sa):
self.sa = sa
def __get__(self, obj, type):
try:
mapper = orm.class_mapper(type)
if mapper:
if type.__module__ == 'pyjeeves.models.raw':
return type.query_class(mapper, session=self.sa.raw_session())
else:
return type.query_class(mapper, session=self.sa.meta_session())
except UnmappedClassError:
return None
class DBConnector(object):
"""This class is used to control the SQLAlchemy integration"""
def __init__(self, enabled_sessions=['raw'], metadata=None):
def __init__(self, enabled_clients=['raw'], metadata=None):
logger.info("Creating engines and sessionmakers")
self.raw_session, self.meta_session = self.create_scoped_session(enabled_sessions)
self.Model = self.make_declarative_base(metadata)
# self.Query = Query
self.raw, self.raw_engine = (self.raw_session() if 'raw' in enabled_clients else {})
self.meta = (self.meta_session() if 'meta' in enabled_clients else {})
@property
def metadata(self):
"""Returns the metadata"""
return self.Model.metadata
def callproc(self, procedure="", params=[]):
conn = self.raw_engine.raw_connection()
# @property
# def _config(self):
# """Returns the configuration"""
# return config()
with conn.cursor() as cursor:
try:
retval = cursor.callproc(procedure, params)
try:
cursor.nextset()
retval = cursor.fetchall()
except OperationalError:
logger.debug("Executed statement has no resultset")
def make_declarative_base(self, metadata=None):
"""Creates the declarative base."""
base = declarative_base(cls=Model, name='Model',
metadata=metadata,
metaclass=MetaBaseModel)
base.query = _QueryProperty(self)
return base
conn.commit()
def create_scoped_session(self, sessions=[]):
RawSession, MetaSession = None, None
if 'raw' in sessions:
raw_engine = create_engine(
'mssql+pymssql://{user}:{pw}@{host}:{port}/{db}?charset=utf8'.format(
**config.config['databases']['raw']),
implicit_returning=False)
finally:
conn.close()
RawSession = scoped_session(sessionmaker(bind=raw_engine))
return retval
if 'meta' in sessions:
meta_engine = create_engine(
'mysql+pymysql://{user}:{pw}@{host}:{port}/{db}?charset=utf8mb4'.format(
**config.config['databases']['meta']))
def execute(self, operation=""):
conn = self.raw_engine.raw_connection()
MetaSession = scoped_session(sessionmaker(bind=meta_engine))
with conn.cursor(as_dict=True) as cursor:
try:
cursor.execute(operation)
results = cursor.fetchall()
finally:
conn.close()
return results
return RawSession, MetaSession
def raw_session(self):
uri = 'mssql+pymssql://{user}:{pw}@{host}:{port}/{db}?charset=utf8'.format(
**config.config['databases']['raw'])
sql_client_config = {'SQL_DATABASE_URI': uri}
db = SQLClient(sql_client_config, query_class=BaseFilterQuery)
return db.session, db.engine
def meta_session(self):
meta_engine = create_engine(
'mysql+pymysql://{user}:{pw}@{host}:{port}/{db}?charset=utf8mb4'.format(
**config.config['databases']['meta']))
return scoped_session(sessionmaker(bind=meta_engine))