API Reference#
The sqlservice package imports commonly used objects into it’s top-level namespace:
from sqlservice import (
AsyncDatabase,
AsyncSession,
Database,
ModelBase,
ModelMeta,
Session,
as_declarative,
declarative_base,
)
Database#
The database module provides a unified class interface to SQLAlchemy engine connection and session objects.
- class sqlservice.database.Database(uri: str, *, model_class: ~typing.Type[~sqlservice.model.ModelBase] | None = None, session_class: ~typing.Type[~sqlservice.session.Session] = <class 'sqlservice.session.Session'>, autoflush: bool | None = None, expire_on_commit: bool | None = None, isolation_level: str | None = None, pool_size: int | None = None, pool_timeout: int | float | None = None, pool_recycle: int | float | None = None, pool_pre_ping: bool | None = None, poolclass: ~typing.Type[~sqlalchemy.pool.base.Pool] | None = None, max_overflow: int | None = None, paramstyle: str | None = None, execution_options: ~typing.Dict[str, ~typing.Any] | None = None, echo: bool | str | None = None, echo_pool: bool | str | None = None, engine_options: ~typing.Dict[str, ~typing.Any] | None = None, session_options: ~typing.Dict[str, ~typing.Any] | None = None)[source]#
Database engine and ORM session management class.
The primary purpose of this class is to provide SQLAlchemy database connections and ORM sessions via a single interface.
Connections and session are provided using the factory methods:
connect()
- Return a new database engine connection object.session()
- Return a new database session object.begin()
- Start a new database session transaction and return session object.
In addition, model metadata operations are available:
create_all()
- Create all database tables defined on base ORM model class.drop_all()
- Drop all database tables defined on declarative base ORM model class.reflect()
- Reflect database schema from database connection.
Dictionary access to the underlying table objects and model classes from the declarative base class are at:
tables
models
Note
This class uses the new 2.0 style SQLAlchemy API. Learn more at SQLAlchemy 1.4 / 2.0 Tutorial.
- begin(*, autoflush: bool | None = None, expire_on_commit: bool | None = None, **kwargs: Any) Generator[Session, None, None] [source]#
Context manager that begins a new session transaction.
Commit and rollback logic will be handled automatically.
Session#
The database session module.
- class sqlservice.session.Session(bind: _SessionBind | None = None, *, autoflush: bool = True, future: Literal[True] = True, expire_on_commit: bool = True, autobegin: bool = True, twophase: bool = False, binds: Dict[_SessionBindKey, _SessionBind] | None = None, enable_baked_queries: bool = True, info: _InfoType | None = None, query_cls: Type[Query[Any]] | None = None, autocommit: Literal[False] = False, join_transaction_mode: JoinTransactionMode = 'conditional_savepoint', close_resets_only: bool | _NoArg = _NoArg.NO_ARG)[source]#
Manages persistence operations for ORM-mapped objects.
See also
For full SQLAlchemy documentation: https://docs.sqlalchemy.org/en/latest/orm/session.html
- all(statement: Executable, params: Mapping[str, Any] | Sequence[Mapping[str, Any]] | None = None, execution_options: Mapping[str, Any] | None = None, bind_arguments: Mapping[str, Any] | None = None) Sequence[Row[Any] | Any] [source]#
Return list of objects from execution of statement.
If statement is a selectable that returns ORM models, then the model objects will be returned.
If a joined-load is used that requires uniquification, then
Result.unique()
will automatically be called.- Parameters:
statement – An executable statement such as
sqlalchemy.select
.params – Optional dictionary or list of dictionaries containing bound parameter values. If a single dictionary, single-row execution occurs; if a list of dictionaries, an “executemany” will be invoked. The keys in each dictionary must correspond to parameter names present in the statement.
execution_options – Optional dictionary of execution options, which will be associated with the statement execution. This dictionary can provide a subset of the options that are accepted by
sqlalchemy._future.Connection.execution_options
, and may also provide additional options understood only in an ORM context.bind_arguments – Dictionary of additional arguments to determine the bind. May include “mapper”, “bind”, or other custom arguments. Contents of this dictionary are passed to the
Session.get_bind
method.
- first(statement: Executable, params: Mapping[str, Any] | Sequence[Mapping[str, Any]] | None = None, execution_options: Mapping[str, Any] | None = None, bind_arguments: Mapping[str, Any] | None = None) Row[Any] | Any | None [source]#
Return first result of statement or
None
if no results.If statement is a selectable that returns ORM models, then the model object will be returned.
- Parameters:
statement – An executable statement such as
sqlalchemy.select
.params – Optional dictionary or list of dictionaries containing bound parameter values. If a single dictionary, single-row execution occurs; if a list of dictionaries, an “executemany” will be invoked. The keys in each dictionary must correspond to parameter names present in the statement.
execution_options – Optional dictionary of execution options, which will be associated with the statement execution. This dictionary can provide a subset of the options that are accepted by
sqlalchemy._future.Connection.execution_options
, and may also provide additional options understood only in an ORM context.bind_arguments – Dictionary of additional arguments to determine the bind. May include “mapper”, “bind”, or other custom arguments. Contents of this dictionary are passed to the
Session.get_bind
method.
- one(statement: Executable, params: Mapping[str, Any] | Sequence[Mapping[str, Any]] | None = None, execution_options: Mapping[str, Any] | None = None, bind_arguments: Mapping[str, Any] | None = None) Row[Any] | Any [source]#
Return exactly one result or raise an exception.
If statement is a selectable that returns ORM models, then the model object will be returned.
- Parameters:
statement – An executable statement such as
sqlalchemy.select
.params – Optional dictionary or list of dictionaries containing bound parameter values. If a single dictionary, single-row execution occurs; if a list of dictionaries, an “executemany” will be invoked. The keys in each dictionary must correspond to parameter names present in the statement.
execution_options – Optional dictionary of execution options, which will be associated with the statement execution. This dictionary can provide a subset of the options that are accepted by
sqlalchemy._future.Connection.execution_options
, and may also provide additional options understood only in an ORM context.bind_arguments – Dictionary of additional arguments to determine the bind. May include “mapper”, “bind”, or other custom arguments. Contents of this dictionary are passed to the
Session.get_bind
method.
- one_or_none(statement: Executable, params: Mapping[str, Any] | Sequence[Mapping[str, Any]] | None = None, execution_options: Mapping[str, Any] | None = None, bind_arguments: Mapping[str, Any] | None = None) Row[Any] | Any | None [source]#
Return exactly one result or
None
if no results or raise if more than one result.If statement is a selectable that returns ORM models, then the model object will be returned.
- Parameters:
statement – An executable statement such as
sqlalchemy.select
.params – Optional dictionary or list of dictionaries containing bound parameter values. If a single dictionary, single-row execution occurs; if a list of dictionaries, an “executemany” will be invoked. The keys in each dictionary must correspond to parameter names present in the statement.
execution_options – Optional dictionary of execution options, which will be associated with the statement execution. This dictionary can provide a subset of the options that are accepted by
sqlalchemy._future.Connection.execution_options
, and may also provide additional options understood only in an ORM context.bind_arguments – Dictionary of additional arguments to determine the bind. May include “mapper”, “bind”, or other custom arguments. Contents of this dictionary are passed to the
Session.get_bind
method.
- save(model: T) T [source]#
Save model in the database using insert, update, or upsert on primary key(s).
See also
See
Session.save_all()
for more details.- Parameters:
model – Models to save to database.
- Raises:
- TypeError – On validation errors.
- sqlalchemy.exc.* – On execution errors.
- save_all(models: Iterable[Any] | Any) List[Any] [source]#
Save models into the database using insert, update, or upsert on primary key(s).
It’s not required that each model be of the same model class. A mixture of model classes are allowed.
The “upsert” will only occur for models that have their primary key(s) set. Upsert on keys other than primary keys is not supported. The “upsert” itself occurs at the application-layer only and does not take advantage of any database specific upsert support. Therefore, it is possible that there could be a race-condition that would result in an
IntegrityError
if a model’s primary key is set, but not found in the database, but is then inserted into the database by another process before this method can insert it.If a corresponding model instance with the same primary key(s) as one of the models exists in the current sessions but not in the models list, then the model in the models list will be merged with the corresponding instance in the session state.
If multiple instances of a model class in models have the same primary-key, an exception will be raised.
Warning
Saving will result in 1 “SELECT” query for every unique model class in the list of models. The maximum returned result from each query would be the total not of entities of each model class in the list of models.
- Parameters:
models – Models to save to database.
- Raises:
- TypeError – On validation errors.
- sqlalchemy.exc.* – On execution errors.
- sqlservice.session.copy_model_pk(from_model: Any, to_model: Any) None [source]#
Transfer primary key value from
parent_model
tochild_model
.
- sqlservice.session.model_pk(model: Any) Tuple[Any, ...] [source]#
Return tuple of primary-key values for given model instance.
- sqlservice.session.pk_filter(*models) ColumnElement[bool] [source]#
Return SQL filter expression over primary-key values of given models.
The filter will have the form:
(pk_col1 = models[0].pk_col1 ... AND pk_colN = models[0].pk_colN) OR (pk_col1 = models[1].pk_col1 ... AND pk_colN = models[1].pk_colN) ... OR (pk_col1 = models[M].pk_col1 ... AND pk_colN = models[M].pk_colN)
Async Database#
The async_database module provides an asyncio version of sqlservice.database.Database
.
- class sqlservice.async_database.AsyncDatabase(uri: str, *, model_class: ~typing.Type[~sqlservice.model.ModelBase] | None = None, session_class: ~typing.Type[~sqlservice.async_session.AsyncSession] = <class 'sqlservice.async_session.AsyncSession'>, autoflush: bool | None = None, expire_on_commit: bool | None = False, isolation_level: str | None = None, pool_size: int | None = None, pool_timeout: int | float | None = None, pool_recycle: int | float | None = None, pool_pre_ping: bool | None = None, poolclass: ~typing.Type[~sqlalchemy.pool.base.Pool] | None = None, max_overflow: int | None = None, paramstyle: str | None = None, execution_options: ~typing.Dict[str, ~typing.Any] | None = None, echo: bool | str | None = None, echo_pool: bool | str | None = None, engine_options: ~typing.Dict[str, ~typing.Any] | None = None, session_options: ~typing.Dict[str, ~typing.Any] | None = None)[source]#
Asynchronous Database engine and ORM session management class.
The primary purpose of this class is to provide SQLAlchemy database connections and ORM sessions via a single interface that is compatible with asyncio.
Connections and session are provided using the factory methods:
connect()
- Return a new database engine connection object.session()
- Return a new database session object.begin()
- Start a new database session transaction and return session object.
In addition, model metadata operations are available:
create_all()
- Create all database tables defined on base ORM model class.drop_all()
- Drop all database tables defined on declarative base ORM model class.reflect()
- Reflect database schema from database connection.
Dictionary access to the underlying table objects and model classes from the declarative base class are at:
tables
models
Note
This class uses the new 2.0 style SQLAlchemy API. Learn more at SQLAlchemy 1.4 / 2.0 Tutorial.
- begin(*, autoflush: bool | None = None, expire_on_commit: bool | None = None, **kwargs: Any) AsyncGenerator[AsyncSession, None] [source]#
Async context manager that begins a new session transaction.
Commit and rollback logic will be handled automatically.
- async create_all(**kwargs: Any) None [source]#
Create all database schema defined in declarative base class.
- create_engine() AsyncEngine [source]#
Return instance of SQLAlchemy async-engine using database settings.
- async drop_all(**kwargs: Any) None [source]#
Drop all database schema defined in declarative base class.
- session(*, autoflush: bool | None = None, expire_on_commit: bool | None = None, **kwargs: Any) AsyncSession [source]#
Return new async-session instance using database settings.
Async Session#
- class sqlservice.async_session.AsyncSession(bind: _AsyncSessionBind | None = None, *, binds: Dict[_SessionBindKey, _AsyncSessionBind] | None = None, sync_session_class: Type[Session] | None = None, **kw: Any)[source]#
Manages persistence operations for ORM-mapped objects using asyncio.
- async all(statement: Executable, params: Mapping[str, Any] | Sequence[Mapping[str, Any]] | None = None, execution_options: Mapping[str, Any] | None = None, bind_arguments: Mapping[str, Any] | None = None) Sequence[Row[Any] | Any] [source]#
Return list of objects from execution of statement.
If statement is a selectable that returns ORM models, then the model objects will be returned.
If a joined-load is used that requires uniquification, then
Result.unique()
will automatically be called.- Parameters:
statement – An executable statement such as
sqlalchemy.select
.params – Optional dictionary or list of dictionaries containing bound parameter values. If a single dictionary, single-row execution occurs; if a list of dictionaries, an “executemany” will be invoked. The keys in each dictionary must correspond to parameter names present in the statement.
execution_options – Optional dictionary of execution options, which will be associated with the statement execution. This dictionary can provide a subset of the options that are accepted by
sqlalchemy._future.Connection.execution_options
, and may also provide additional options understood only in an ORM context.bind_arguments – Dictionary of additional arguments to determine the bind. May include “mapper”, “bind”, or other custom arguments. Contents of this dictionary are passed to the
Session.get_bind
method.
- async first(statement: Executable, params: Mapping[str, Any] | Sequence[Mapping[str, Any]] | None = None, execution_options: Mapping[str, Any] | None = None, bind_arguments: Mapping[str, Any] | None = None) Row[Any] | Any | None [source]#
Return first result of statement or
None
if no results.If statement is a selectable that returns ORM models, then the model object will be returned.
- Parameters:
statement – An executable statement such as
sqlalchemy.select
.params – Optional dictionary or list of dictionaries containing bound parameter values. If a single dictionary, single-row execution occurs; if a list of dictionaries, an “executemany” will be invoked. The keys in each dictionary must correspond to parameter names present in the statement.
execution_options – Optional dictionary of execution options, which will be associated with the statement execution. This dictionary can provide a subset of the options that are accepted by
sqlalchemy._future.Connection.execution_options
, and may also provide additional options understood only in an ORM context.bind_arguments – Dictionary of additional arguments to determine the bind. May include “mapper”, “bind”, or other custom arguments. Contents of this dictionary are passed to the
Session.get_bind
method.
- async one(statement: Executable, params: Mapping[str, Any] | Sequence[Mapping[str, Any]] | None = None, execution_options: Mapping[str, Any] | None = None, bind_arguments: Mapping[str, Any] | None = None) Row[Any] | Any [source]#
Return exactly one result or raise an exception.
If statement is a selectable that returns ORM models, then the model object will be returned.
- Parameters:
statement – An executable statement such as
sqlalchemy.select
.params – Optional dictionary or list of dictionaries containing bound parameter values. If a single dictionary, single-row execution occurs; if a list of dictionaries, an “executemany” will be invoked. The keys in each dictionary must correspond to parameter names present in the statement.
execution_options – Optional dictionary of execution options, which will be associated with the statement execution. This dictionary can provide a subset of the options that are accepted by
sqlalchemy._future.Connection.execution_options
, and may also provide additional options understood only in an ORM context.bind_arguments – Dictionary of additional arguments to determine the bind. May include “mapper”, “bind”, or other custom arguments. Contents of this dictionary are passed to the
Session.get_bind
method.
- async one_or_none(statement: Executable, params: Mapping[str, Any] | Sequence[Mapping[str, Any]] | None = None, execution_options: Mapping[str, Any] | None = None, bind_arguments: Mapping[str, Any] | None = None) Row[Any] | Any | None [source]#
Return exactly one result or
None
if no results or raise if more than one result.If statement is a selectable that returns ORM models, then the model object will be returned.
- Parameters:
statement – An executable statement such as
sqlalchemy.select
.params – Optional dictionary or list of dictionaries containing bound parameter values. If a single dictionary, single-row execution occurs; if a list of dictionaries, an “executemany” will be invoked. The keys in each dictionary must correspond to parameter names present in the statement.
execution_options – Optional dictionary of execution options, which will be associated with the statement execution. This dictionary can provide a subset of the options that are accepted by
sqlalchemy._future.Connection.execution_options
, and may also provide additional options understood only in an ORM context.bind_arguments – Dictionary of additional arguments to determine the bind. May include “mapper”, “bind”, or other custom arguments. Contents of this dictionary are passed to the
Session.get_bind
method.
- async save(model: T) T [source]#
Save model in the database using insert, update, or upsert on primary key(s).
See also
See
Session.save_all()
for more details.- Parameters:
model – Models to save to database.
- Raises:
- TypeError – On validation errors.
- sqlalchemy.exc.* – On execution errors.
- async save_all(models: Iterable[Any] | Any) List[Any] [source]#
Save models into the database using insert, update, or upsert on primary key(s).
It’s not required that each model be of the same model class. A mixture of model classes are allowed.
The “upsert” will only occur for models that have their primary key(s) set. Upsert on keys other than primary keys is not supported. The “upsert” itself occurs at the application-layer only and does not take advantage of any database specific upsert support. Therefore, it is possible that there could be a race-condition that would result in an
IntegrityError
if a model’s primary key is set, but not found in the database, but is then inserted into the database by another process before this method can insert it.If a corresponding model instance with the same primary key(s) as one of the models exists in the current sessions but not in the models list, then the model in the models list will be merged with the corresponding instance in the session state.
If multiple instances of a model class in models have the same primary-key, an exception will be raised.
Warning
Saving will result in 1 “SELECT” query for every unique model class in the list of models. The maximum returned result from each query would be the total not of entities of each model class in the list of models.
- Parameters:
models – Models to save to database.
- Raises:
- TypeError – On validation errors.
- sqlalchemy.exc.* – On execution errors.
Model#
The declarative base model class for SQLAlchemy ORM.
- class sqlservice.model.ModelBase(**kwargs: Any)[source]#
Declarative base for all ORM model classes.
- classmethod delete() Delete [source]#
Return instance of
sqlalchemy.delete(Model)
for use in querying.
- classmethod insert() Insert [source]#
Return instance of
sqlalchemy.insert(Model)
for use in querying.
- classmethod select() Select [source]#
Return instance of
sqlalchemy.select(Model)
for use in querying.
- to_dict(*, exclude_relationships: bool = False, lazyload: bool = False) Dict[str, Any] [source]#
Serialize ORM loaded data to dictionary.
Only the loaded data, i.e. data previously fetched from the database, will be serialized. Lazy-loaded columns and relationships will be excluded to avoid extra database queries.
By default, only table columns will be included. To include relationship fields, set
include_relationships=True
. This will nestto_dict()
calls to the relationship models.
- class sqlservice.model.ModelMeta(name, bases, dct)[source]#
Model metaclass that prepares model classes for event registration hooks.
- sqlservice.model.as_declarative(*, metadata: ~sqlalchemy.sql.schema.MetaData | None = None, metaclass: ~typing.Type[~sqlalchemy.orm.decl_api.DeclarativeMeta] | None = <class 'sqlservice.model.ModelMeta'>, **kwargs: ~typing.Any) Callable[[Type[ModelBase]], Type[ModelBase]] [source]#
Decorator version of
declarative_base()
.
- sqlservice.model.declarative_base(cls: ~typing.Type[~sqlservice.model.ModelBase] = <class 'sqlservice.model.ModelBase'>, *, metadata: ~sqlalchemy.sql.schema.MetaData | None = None, metaclass: ~typing.Type[~sqlalchemy.orm.decl_api.DeclarativeMeta] | None = None, **kwargs: ~typing.Any) Type[ModelBase] [source]#
Function that converts a normal class into a SQLAlchemy declarative base class.
- Parameters:
cls – A type to use as the base for the generated declarative base class. May be a class or tuple of classes. Defaults to
ModelBase
.metadata – An optional MetaData instance. All Table objects implicitly declared by subclasses of the base will share this MetaData. A MetaData instance will be created if none is provided. Defaults to
None
which will associate a new metadata instance with the returned declarative base class.metaclass – A metaclass or
__metaclass__
compatible callable to use as the meta type of the generated declarative base class. Defaults toModelMeta
.
- Keyword Arguments:
sqlalchemy.ext.declarative.declarative_base. (All other keyword arguments are passed to) –
Event#
The event module with declarative ORM event decorators and event registration.
- class sqlservice.event.AttributeEventDecorator(attribute: Any, **event_kwargs: Any)[source]#
Base class for attribute event decorators.
- class sqlservice.event.Event(name, attribute, listener, kwargs)[source]#
Universal event class used when registering events.
- class sqlservice.event.EventDecorator(**event_kwargs: Any)[source]#
Base class for event decorators that attaches metadata to function object so that
register()
can find the event definition.
- class sqlservice.event.MapperEventDecorator(**event_kwargs: Any)[source]#
Base class for mapper event decorators.
- class sqlservice.event.after_delete(**event_kwargs: Any)[source]#
Event decorator for the
after_delete
event.
- class sqlservice.event.after_insert(**event_kwargs: Any)[source]#
Event decorator for the
after_insert
event.
- class sqlservice.event.after_save(**event_kwargs: Any)[source]#
Event decorator for the
after_insert
andafter_update
events.
- class sqlservice.event.after_update(**event_kwargs: Any)[source]#
Event decorator for the
after_update
event.
- class sqlservice.event.before_delete(**event_kwargs: Any)[source]#
Event decorator for the
before_delete
event.
- class sqlservice.event.before_insert(**event_kwargs: Any)[source]#
Event decorator for the
before_insert
event.
- class sqlservice.event.before_save(**event_kwargs: Any)[source]#
Event decorator for the
before_insert
andbefore_update
events.
- class sqlservice.event.before_update(**event_kwargs: Any)[source]#
Event decorator for the
before_update
event.
- class sqlservice.event.on_append(attribute: Any, **event_kwargs: Any)[source]#
Event decorator for the
append
event.
- class sqlservice.event.on_bulk_replace(attribute: Any, **event_kwargs: Any)[source]#
Event decorator for the
bulk_replace
event.
- class sqlservice.event.on_dispose_collection(attribute: Any, **event_kwargs: Any)[source]#
Event decorator for the
dispose_collection
event.
- class sqlservice.event.on_expire(**event_kwargs: Any)[source]#
Event decorator for the
expire
event.
- class sqlservice.event.on_init_collection(attribute: Any, **event_kwargs: Any)[source]#
Event decorator for the
init_collection
event.
- class sqlservice.event.on_init_scalar(attribute: Any, **event_kwargs: Any)[source]#
Event decorator for the
init_scalar
event.
- class sqlservice.event.on_modified(attribute: Any, **event_kwargs: Any)[source]#
Event decorator for the
modified
event.
- class sqlservice.event.on_refresh(**event_kwargs: Any)[source]#
Event decorator for the
refresh
event.
- class sqlservice.event.on_remove(attribute: Any, **event_kwargs: Any)[source]#
Event decorator for the
remove
event.