object_session
is a callable within the sqlalchemy.orm.session
module of the SQLAlchemy project.
Session
is another callable from the sqlalchemy.orm.session
package with code examples.
sqlalchemy-utils (project documentation and PyPI package information) is a code library with various helper functions and new data types that make it easier to use SQLAlchemy when building projects that involve more specific storage requirements such as currency. The wide array of data types includes ranged values and aggregated attributes.
sqlalchemy-utils / sqlalchemy_utils / functions / orm.py
# orm.py
from collections import OrderedDict
from functools import partial
from inspect import isclass
from operator import attrgetter
import six
import sqlalchemy as sa
from sqlalchemy.engine.interfaces import Dialect
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import mapperlib
from sqlalchemy.orm.attributes import InstrumentedAttribute
from sqlalchemy.orm.exc import UnmappedInstanceError
from sqlalchemy.orm.properties import ColumnProperty, RelationshipProperty
from sqlalchemy.orm.query import _ColumnEntity
from sqlalchemy.orm.session import object_session
from sqlalchemy.orm.util import AliasedInsp
from ..utils import is_sequence
def get_class_by_table(base, table, data=None):
found_classes = set(
c for c in base._decl_class_registry.values()
if hasattr(c, '__table__') and c.__table__ is table
)
if len(found_classes) > 1:
if not data:
raise ValueError(
"Multiple declarative classes found for table '{0}'. "
"Please provide data parameter for this function to be able "
"to determine polymorphic scenarios.".format(
table.name
)
)
else:
for cls in found_classes:
mapper = sa.inspect(cls)
polymorphic_on = mapper.polymorphic_on.name
if polymorphic_on in data:
## ... source file abbreviated to get to object_session examples ...
if isinstance(mixed, sa.Table):
mappers = [
mapper for mapper in mapperlib._mapper_registry
if mixed in mapper.tables
]
if len(mappers) > 1:
raise ValueError(
"Multiple mappers found for table '%s'." % mixed.name
)
elif not mappers:
raise ValueError(
"Could not get mapper for table '%s'." % mixed.name
)
else:
return mappers[0]
if not isclass(mixed):
mixed = type(mixed)
return sa.inspect(mixed)
def get_bind(obj):
if hasattr(obj, 'bind'):
conn = obj.bind
else:
try:
conn = object_session(obj).bind
except UnmappedInstanceError:
conn = obj
if not hasattr(conn, 'execute'):
raise TypeError(
'This method accepts only Session, Engine, Connection and '
'declarative model objects.'
)
return conn
def get_primary_keys(mixed):
return OrderedDict(
(
(key, column) for key, column in get_columns(mixed).items()
if column.primary_key
)
)
def get_tables(mixed):
if isinstance(mixed, sa.Table):
return [mixed]
elif isinstance(mixed, sa.Column):
## ... source file continues with no further object_session examples...