"""
Fictive Kin library for using `SQLAlchemy <https://www.sqlalchemy.org/>`_
"""
from argparse import Namespace
import functools
import weakref
import sqlalchemy
from sqlalchemy.ext import declarative
from sqlalchemy.orm.exc import MultipleResultsFound
[docs]class ModelDescriptor(object):
# pylint: disable=too-few-public-methods
"""
Lazy reference to a `declarative_base` subclass
In large applications, "models" are often declared in separate
files from each other and from "controllers" or "views" that
utilize the models. This can easily lead to "circular import"
issues in Python that can require substantial refactoring or
subtle non-idiomatc hacks of Python's import behavior.
`SQLAlchemy`_ provides some features that can alleviate this concern,
most notably the
`Declarative <https://docs.sqlalchemy.org/en/13/orm/extensions/declarative/index.html>`_
extension that, among other features, allows strings to be used in
place of class references. E.g.::
from sqlalchemy import Column, Integer, ForeignKey
from sqlalchemy.orm import relationship
from application.models import Base
class ChildModel(Base):
__tablename__ = 'child'
id = Column(Integer, primary_key=True)
parent_id = Column(ForeignKey('parent.id'))
#: We don't need to `from parents import ParentModel`
#: here; the string reference will be resolved at run time
parent = relationship('ParentModel')
This descriptor uses the `Declarative`_ extension to wrap a string
reference to a `declarative_base` subclass (i.e., a model class)
in a generalized way. When the owning class calls the `__get__`
method, the descriptor will automatically dereference the string
reference and return the indicated `declarative_base` subclass.
E.g.:
.. testsetup:: modeldscriptor
from sqlalchemy import Column, Integer
from sqlalchemy.ext.declarative import declarative_base
from fictive.sqlalchemy import ModelDescriptor
Base = declarative_base()
class ExampleModel(Base):
__tablename__ = 'example_model_table'
id = Column(Integer, primary_key=True)
.. testcode:: modeldscriptor
class MyController(object):
#: We don't need to `from models import ExampleModel`
#: here; the string reference will be resolved at run time
model_class = ModelDescriptor('ExampleModel', Base)
def describe_model(self):
return self.model_class.__tablename__
.. doctest:: modeldscriptor
>>> MyController().describe_model()
'example_model_table'
"""
[docs] def __init__(self, model: str, base: declarative.api.DeclarativeMeta):
"""
:param model:
the "string name" of model class. The model class
**MUST** be a subclass of `base`
:type model: str
:param base:
a base class created by
`sqlalchemy.ext.declarative.declarative_base`
:type base: `sqlalchemy.ext.declarative.api.DeclarativeMeta`
"""
self.model = model
self.base = weakref.proxy(base)
def __get__(self, instance, owner):
"""
dereferences the "string name" to obtain the model class
"""
if isinstance(self.model, str):
prop = Namespace(parent=instance)
self.model = declarative.clsregistry._resolver(
self.base, prop)[1](self.model)()
return self.model
[docs]def find_or_create(
session: sqlalchemy.orm.session.Session,
model_class: declarative.api.DeclarativeMeta,
**kwargs):
"""
retrieves an instance from the database or creates one if it doesn't exist
E.g.:
.. testsetup:: findorcreate
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
engine = create_engine('sqlite:///:memory:')
Session = sessionmaker(bind=engine)
session = Session()
DeclarativeBase = declarative_base()
.. testcode:: findorcreate
from sqlalchemy import Column, Integer, String
from fictive.sqlalchemy import find_or_create
class Model(DeclarativeBase):
__tablename__ = 'model'
id = Column(Integer, primary_key=True)
value = Column(String)
DeclarativeBase.metadata.create_all(engine)
.. doctest:: findorcreate
>>> first = Model(value='first')
>>> session.add(first)
>>> session.commit()
>>> found = find_or_create(session, Model, value='first')
>>> (found.id, found.value)
(1, 'first')
>>> found is first
True
>>> created = find_or_create(session, Model, value='second')
>>> (created.id, created.value)
(None, 'second')
>>> created in session
False
.. testcleanup:: findorcreate
DeclarativeBase.metadata.drop_all(engine)
session.close()
engine.dispose()
:param session: an active SQLAlchemy session
:type session: `sqlalchemy.orm.session.Session`
:param model_class: the model class to find or create
:type model_class: `sqlalchemy.ext.declarative.api.DeclarativeMeta`
:param **kwargs:
these keyword arguments will be passed to
`sqlalchemy.orm.query.Query.filter_by`
:param __require_unique:
require the filter criteria produce a unique result. If
`False` (the default), will return the first matching record
if any exist. If `True` and more than one matching record
exists, raises `sqlalchemy.orm.exc.MultipleResultsFound`
:type __require_unique: `bool`
:param bool __suppress_errors:
suppress any SQLAlchemy ORM errors. The default is `False`.
If this is set to `True`, any query errors (other than
`sqlalchemy.orm.exc.MultipleResultsFound`) will be handled as
if no matching record had been found
:type __suppress_errors: `bool`
"""
require_unique = kwargs.pop('__require_unique', False)
suppress_errors = kwargs.pop('__suppress_errors', False)
try:
filtered_query = session.query(model_class).filter_by(**kwargs)
if require_unique:
instance = filtered_query.one_or_none()
else:
instance = filtered_query.first()
except MultipleResultsFound:
# would only have been raised if `require_unique` is `True`
raise
except sqlalchemy.exc.SQLAlchemyError:
if not suppress_errors:
raise
return model_class(**kwargs)
return instance or model_class(**kwargs)
[docs]class FindOrCreateDescriptor(object):
"""
Descriptor for attaching a `find_or_create` method to a `model` class
.. testsetup:: findorcreatedescriptor
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
engine = create_engine('sqlite:///:memory:')
Session = sessionmaker(bind=engine)
session = Session()
session_factory = Session
DeclarativeBase = declarative_base()
.. testcode:: findorcreatedescriptor
from fictive.sqlalchemy import FindOrCreateDescriptor
class Model(DeclarativeBase):
__tablename__ = 'model'
id = Column(Integer, primary_key=True)
value = Column(String)
find_or_create = FindOrCreateDescriptor(session_factory)
DeclarativeBase.metadata.create_all(engine)
.. doctest:: findorcreatedescriptor
>>> created = Model.find_or_create(value='created')
>>> session.add(created)
>>> session.commit()
>>> (created.id, created.value)
(1, 'created')
>>> retrieved = Model.find_or_create(value='created')
>>> (retrieved.id, retrieved.value)
(1, 'created')
>>> # each call to find_or_create falls the session_factory so
>>> # the instances may be in different sessions unless the
>>> # factory returns the same session over multiple calls
>>> # (e.g., `sqlalchemy.orm.scoping.scoped_session`)
>>> retrieved is created
False
.. testcleanup:: findorcreatedescriptor
DeclarativeBase.metadata.drop_all(engine)
session.close()
engine.dispose()
"""
# pylint: disable=too-few-public-methods
[docs] def __init__(self, session_factory):
self.session_factory = session_factory
def __get__(self, obj, owner):
session = self.session_factory()
partial = functools.partial(find_or_create, session, owner)
return functools.wraps(find_or_create)(partial)