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.

close() None[source]#

Close engine connection.

connect() Connection[source]#

Return new connection instance using database settings.

create_all(**kwargs: Any) None[source]#

Create all database schema defined in declarative base class.

create_engine() Engine[source]#

Return instance of SQLAlchemy engine using database settings.

drop_all(**kwargs: Any) None[source]#

Drop all database schema defined in declarative base class.

ping() bool[source]#

Return whether database can be accessed.

reflect(**kwargs: Any) None[source]#

Reflect database schema from database connection.

session(*, autoflush: bool | None = None, expire_on_commit: bool | None = None, **kwargs: Any) Session[source]#

Return new session instance using database settings.

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 to child_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 close() None[source]#

Close engine connection.

connect() AsyncConnection[source]#

Return new async-connection instance using database settings.

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.

async ping() bool[source]#

Return whether database can be accessed.

async reflect(**kwargs: Any) None[source]#

Reflect database schema from database connection.

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.

pk() Tuple[Any, ...][source]#

Return primary key identity of model instance.

classmethod select() Select[source]#

Return instance of sqlalchemy.select(Model) for use in querying.

set(**kwargs: Any) None[source]#

Update model using keyword arguments.

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 nest to_dict() calls to the relationship models.

classmethod update() Update[source]#

Return instance of sqlalchemy.update(Model) for use in querying.

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 to ModelMeta.

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 and after_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 and before_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_load(**event_kwargs: Any)[source]#

Event decorator for the load 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.

class sqlservice.event.on_set(attribute: Any, **event_kwargs: Any)[source]#

Event decorator for the set event.

sqlservice.event.register(cls: type, dct: dict) None[source]#

Register events defined on a class during metaclass creation.