rwskit.sqlalchemy

Utilities for working with SqlAlchemy.

Attributes

`log`
`B`	A type extending `DeclarativeBase`.
`M`
`TableArgs`	The type of the `__table_args__` attribute on a `Table`.
`DtoModel`	The base type for a DTO model.
`SqlOperator`	The supported SQL operators for use in an `SqlBinaryExpression`.
`ConflictResolutionStrategy`
`Walkable`
`WalkPredicate`

Classes

`AlchemyEngine`	A wrapper around `sqlalchemy.engine.Engine` that provides additional functionality.
`BaseModel`	A base class for creating declarative SqlAlchemy models.
`SqlBinaryExpression`	A class that represents the basic binary expression for an SQL column.
`SqlSelectionCriteria`	A class that represents a conjunction of SqlBinaryExpression.
`SqlOrderExpression`	A class that represents a basic order expression for an SQL column.
`SqlOrderCriteria`	A class that represents a list of `SqlOrderExpressions`.
`DatabaseConnectionConfig`	The options necessary to connect to a database using SqlAlchemy.
`Repository`	A class implementing the basic CRUD operations for the data access layer.

Functions

handle_after_mapper_configured(mapper, cls)

Validate the configuration and create an index on the natural key.

Module Contents

rwskit.sqlalchemy.log[source]

rwskit.sqlalchemy.B[source]: A type extending DeclarativeBase.

rwskit.sqlalchemy.M[source]

rwskit.sqlalchemy.TableArgs[source]: The type of the __table_args__ attribute on a Table.

rwskit.sqlalchemy.DtoModel[source]: The base type for a DTO model.

rwskit.sqlalchemy.SqlOperator[source]: The supported SQL operators for use in an SqlBinaryExpression.

rwskit.sqlalchemy.ConflictResolutionStrategy[source]

rwskit.sqlalchemy.Walkable[source]

rwskit.sqlalchemy.WalkPredicate[source]

class rwskit.sqlalchemy.AlchemyEngine(engine_or_config: sqlalchemy.ext.asyncio.AsyncEngine | sqlalchemy.Engine | DatabaseConnectionConfig, base_model: Type[sqlalchemy.orm.DeclarativeBase], **engine_kwargs: Any)[source]

A wrapper around sqlalchemy.engine.Engine that provides additional functionality.

base_model[source]

property dialect: str[source]: Returns the dialect name of the engine.

property supports_async: bool[source]: Returns true if the engine supports async mode.

property sync_engine: sqlalchemy.Engine[source]: Get a reference to a synchronous sqlalchemy.engine.Engine.

property async_engine: sqlalchemy.ext.asyncio.AsyncEngine[source]

Get a reference to an asynchronous sqlalchemy.ext.asyncio.AsyncEngine.

Returns:: The engine.
Return type:: AsyncEngine
Raises:: ValueError – If the engine does not support async mode.

make_session() → sqlalchemy.Session[source]

Create a session from the AlchemyEngine.session_factory.

..note::: You probably want to use AlchemyEngine.session_scope(), but there may be cases where this is more appropriate.

Returns:: A Session object.
Return type:: Session

make_connection() → sqlalchemy.Connection[source]

Get a connection to the engine.

Returns:: An sqlalchemy.Connection.
Return type:: Connection

make_raw_connection() → sqlalchemy.PoolProxiedConnection[source]

Get a raw connection to the engine.

Note

See Working with Driver SQL and Raw DBAPI Connections in the SqlAlchemy documentation for the difference between regular and raw connections.

Returns:: A sqlalchemy.PoolProxiedConnection.
Return type:: PoolProxiedConnection

session_scope(expire_on_commit: bool = False) → sqlalchemy.Session[source]

A context manager for committing successful transactions when the session is complete or rolling back if there was an exception.

Parameters:: expire_on_commit (bool, default=False) – If True model objects will be marked as stale when the next commit. This will invalidate all relationships and raise an exception if they are accessed outside the session.
Returns:: An sqlalchemy.orm.Session.
Return type:: Session

async async_session_scope(expire_on_commit: bool = False) → sqlalchemy.Session[source]

A context manager for committing successful transactions when the session is complete or rolling back if there was an exception.

Parameters:: expire_on_commit (bool, default=False) – If True model objects will be marked as stale when the next commit. This will invalidate all relationships and raise an exception if they are accessed outside the session.
Returns:: An sqlalchemy.orm.Session.
Return type:: Session

test_scope(expire_on_commit: bool = False) → sqlalchemy.Session[source]

A session scope for testing.

This session will always roll back after exiting the context manager and should not persist any changes to the database.

Parameters:: expire_on_commit (bool, default=False) – If True model objects will be marked as stale when the next commit. This will invalidate all relationships and raise an exception if they are accessed outside the session.
Returns:: An sqlalchemy.orm.Session.
Return type:: Session
Raises:: RuntimeError – If the user tries to commit changes during the session.

async async_test_scope(expire_on_commit: bool = False) → sqlalchemy.Session[source]

A session scope for testing.

This session will always roll back after exiting the context manager and should not persist any changes to the database.

Parameters:: expire_on_commit (bool, default=False) – If True model objects will be marked as stale when the next commit. This will invalidate all relationships and raise an exception if they are accessed outside the session.
Returns:: An sqlalchemy.orm.Session.
Return type:: Session
Raises:: RuntimeError – If the user tries to commit changes during the session.

rwskit.sqlalchemy.handle_after_mapper_configured(mapper: sqlalchemy.orm.Mapper, cls: Type)[source]: Validate the configuration and create an index on the natural key.

class rwskit.sqlalchemy.BaseModel[source]

Bases: sqlalchemy.orm.DeclarativeBase, sqlalchemy.orm.MappedAsDataclass

A base class for creating declarative SqlAlchemy models.

Features

Table Lookup

Find any model derived from this base class by their table name.

Merging Table Args

SqlAlchemy does not merge __table_args__ during inheritance. For example, if you have a base class that will set the schema for all child classes, it will not work if the child class defines its own __table_args__ (e.g., to create a multi-column index). This base class provides a function to merge the __table_args__ of parents with their children.

This funcitonality is enabled by defining __table_args__ as a @declared_attr.directive on the class and returning the value of merge_table_args(). merge_table_args accepts one optional parameter, which can be a tuple or dictionary (the expected types of __table_args__) and will merge these with the table args of its ancestors. In addition to, or alternatively, the merge_table_args method will also look for table arguments in a class attribute named __custom_table_args__.

>>> class Parent(BaseModel):
>>>     @orm.declared_attr.directive
>>>     @classmethod
>>>     def __table_args__(cls):
>>>         return {"schema": "my_schema"}
>>>
>>> class Child(Parent):
>>>     __custom_table_args__ = {"schema": "my_schema"}
>>>
>>>     @orm.declared_attr.directive
>>>     @classmethod
>>>     def __table_args__(cls):
>>>         child_table_args = (
>>>             Index("child_index_name", "column_1", "column_2"),
>>>         )
>>>         return cls.merge_table_args(child_table_args)
>>>
>>>     column_1: orm.Mapped[int]
>>>     column_2: orm.Mapped[int]

Natural Keys

Classes derived from this model must define a natural key. Natural keys are specified by explicitly setting hash=True on a mapped_column. A natural key is intended to identify a set of attributes that uniquely identify a row in the table. The key will be used to define __hash__ and __eq__ for the class. Additionally, a non-unique index will be created on the natural key columns.

Serializable to a Dictionary

There is currently a bug representing models using dataclasses.asdict when the model inherits from MappedAsDataclass and contains a relationship with back_populate defined.

See: https://github.com/sqlalchemy/sqlalchemy/issues/9785

This class provides methods for converting the model to and from dictionaries. This has been tested for several common use cases, but may not be robust for more complex models.

Data Transfer Objects

Any class derived from this base class can build a corresponding DTO class type using the to_dto_class() method. The created DTO class derives from pydantic.BaseModel, which is convenient for offline use and data transfers, for example with FastAPI.

In addition to mirroring the columns, composites, hybrid_properties, and relationships, the DTO object also provides a pretty_print method to format the string representation of the object. It takes one optional parameter line_length.

A DTO instance can be constructed from a :class:BaseModel instance using either the model_validate classmethod method on the DTO or from the to_dto() method of the BaseModel instance.

To exclude an attribute (column, composite, hybrid, relationship, etc.) add dto=False` to the ``info dictionary of the attribute.

Known Limitations

When converting to DTOs the following limitations apply:

A composite column must be a dataclass.
Only regular columns (e.g., no composite columns) can be used as natural keys.
Only the parent side of a relationship is added to the DTO. Namely, a reference to the child and children will be included in the parent, but a reference to the parent will not be included in the child or children. The parent is determined by the presence of foreign keys. For OtO and OtM relationships the parent is the model that does not contain a foreign key. For MtM relationships the parent is determined by looking at the first column of the association table. If all the foreign keys of that column are in the relationship’s local columns then that model is considered the parent. This may cause problems if you directly select the child objects, because their parents will not be loaded into the DTO object.
In general you cannot convert a model to a dictionary or DTO and then back to the exact same model. The from_dict() method does not handle cyclic relationships and will typically not be able to associate a parent instance from the child.

__abstract__ = True[source]

metadata: sqlalchemy.MetaData[source]

hasher[source]: The hasher used to calculate the has value of objects.

__hash__() → int[source]

__eq__(other: Any) → bool[source]

__tablename__()[source]

classmethod merge_table_args(new_args: TableArgs = ()) → TableArgs[source]

This method is intended to be called from __table_args__ when used as a @declared_attr. It will merge the new_args with the arguments of its ancestors. You can also specify additional table arguments in the class variable __custom_table_args__, which will also be merged.

Parameters:: new_args (TableArgs) – Additional table arguments to be merged with the arguments of our ancestors.
Returns:: The merged table arguments as a tuple. The first N arguments of the tuple contain positional arguments passed to the constructor of sa.Table`. If the last element is a dictionary, then it is the keyword arguments passed to the Table constructor, otherwise it is the final positional argument.
Return type:: TableArgs

Examples

>>> class Parent(BaseModel):
>>>     @orm.declared_attr.directive
>>>     @classmethod
>>>     def __table_args__(cls):
>>>         return {"schema": "my_schema"}
>>>
>>> class Child(Parent):
>>>     __custom_table_args__ = {"schema": "my_schema"}
>>>
>>>     @orm.declared_attr.directive
>>>     @classmethod
>>>     def __table_args__(cls):
>>>         child_table_args = (
>>>             Index("child_index_name", "column_1", "column_2"),
>>>         )
>>>         return cls.merge_table_args(child_table_args)
>>>
>>>     column_1: orm.Mapped[int]
>>>     column_2: orm.Mapped[int]

classmethod find_by_table_name(table_name: str) → Type[BaseModel] | None[source]

Find a model derived from this class by its table name.

Parameters:: table_name (str) – The name of the table whose model class you want to find.
Returns:: Returns the model class if the table is found, otherwise None.
Return type:: Type[FindByNameBase], optional

property primary_key: tuple[tuple[str, Any], Ellipsis][source]

Get the primary key of this instance.

The primary key is the set of key/value pairs corresponding to the configured primary key columns. For efficiency reasons, the result is returned as a tuple of tuples.

property natural_key: tuple[tuple[str, Any | None], Ellipsis][source]

Get the natural key of this instance.

The natural key is the set of key/value pairs that uniquely identify this instance. The keys are the names of the columns returned by natural_key_columns() and the values are the current value of the instance. For efficiency when used by the hash function the pairs are returned as a tuple of tuples.

as_table_dict() → dict[str, Any][source]

Returns the table data of the instance as a dictionary.

..warning::: This only works for simple models that have a one to one mapping from mapped columns to table columns. For example, it will break for models containing composite columns.

Returns:: The keys of the dictionary are the table column names and the values are the corresponding values from the instance.
Return type:: dict[str, Any]

to_dict(exclude_attributes: Iterable[str] = (), drop_none: bool = False) → dict[str, Any][source]

Covert the instance to a dictionary.

This method represents the model as a dictionary as it is defined by the class mapping (as opposed to the table mapping).

Note

As of 01/01/2025 calling dataclasses.asdict will raise a recursion error on models that have relationships with back_populate defined.

Note

exclude_attributes does a simple string match on any attribute name in the model and any of its children.

See: https://github.com/sqlalchemy/sqlalchemy/issues/9785

classmethod from_dict(data: dict) → BaseModel[source]

Creates an instance of the class from a dictionary representation of the data, allowing for nested structures and relationships between models. This method also ensures that instances with shared natural keys are not duplicated.

Parameters:: data (dict) – A dictionary containing the attributes and nested relationships of the model. The keys should correspond to the field names of the model, and values should represent their corresponding data. For nested relationships, values are also expected to be dictionaries (for single relationships) or lists of dictionaries (for collections).
Returns:: An instance of the BaseModel subclass created from the dictionary input.
Return type:: BaseModel

classmethod dto_module() → types.ModuleType[source]: Get the module where the DTO class is defined.

classmethod dto_module_name() → str[source]: Get the name of the module where the DTO class is defined.

classmethod dto_class_name() → str[source]: Get the name of the DTO class.

classmethod dto_import_path() → str[source]: Get the import path of the DTO class.

classmethod dto_exclude_attributes() → set[str][source]: A set of attribute names to exclude from the DTO representation.

classmethod to_dto_class() → Type[pydantic.BaseModel][source]

Create a Data Transfer Object (DTO) class that corresponds with this model.

The DTO class will have the same name as the model class but end with the prefix Dto. So, if the model name is MyModel the DTO class will be named MyModelDto.

The DTO class inherits from the pydantic.BaseModel class and contains fields for all regular, composites, hybrid properties, and relationships of this model.

..note::: Regardless of how the SqlAlchemy model class is configured, all primary and foreign key columns are treated as optional and included in the __init__ method with a default value of None. The same applies to database generated integer columns.
..note::: To exclude an attribute from the DTO, set dto=False in the info dictionary of the attribute when defining the model.

Returns:: The DTO type.
Return type:: Type[pydantic.BaseModel]

to_dto() → pydantic.BaseModel[source]

Convert this model to a DTO instance.

Returns:: The equivalent DTO instance.
Return type:: pydantic.BaseModel
Raises:: pydantic.ValidationError – If any of the required fields are not present. Note: this includes database generated fields, such as ids.

classmethod from_dto(dto: pydantic.BaseModel) → M[source]: Create a model instance from a DTO instance.

walk_children(callback: Callable[[B], None], traverse_viewonly: bool = True)[source]

A method to traverse the relationships of a given instance and apply a callback to each node in the traversal.

Parameters:

callback (Callable[[DeclarativeBase], None]) – The function to call on each traversed node.
traverse_viewonly (bool, default=True) – Whether to traverse viewonly relationships.

copy() → M[source]

Return a deep copy of the instance that is not associated in any way with this instance. For example, the new instance is not added to a session when the original item is (which appears to happen if you use copy.deepcopy or dataclasses.replace.

Returns:: A copy of this instance.
Return type:: BaseModel

classmethod class_name() → str[source]: Get the name of the class including the package and module prefix.

classmethod fields() → dict[str, dataclasses.Field][source]: Return a mapping of the field names to their corresponding Field objects. Note, the dictionary is ordered by the field name.

classmethod primary_key_columns() → dict[str, sqlalchemy.orm.ColumnProperty][source]: Return the list of primary key columns of this class.

classmethod natural_key_columns() → dict[str, sqlalchemy.orm.ColumnProperty][source]: Return the list of natural key columns of this class.

classmethod regular_columns() → dict[str, sqlalchemy.orm.ColumnProperty][source]: Return a mapping from an attribute name to its ColumnProperty instance for all regular columns of this class. A regular column is a column listed in Mapper.column_attrs, maps to only one column that is an instance of sa.Column.

classmethod column_properties() → dict[str, sqlalchemy.orm.ColumnProperty][source]: Return a mapping from an attribute name to its ColumnProperty instance for all column properties of this class. A column property is a mapped attribute configured using column_property.

classmethod composite_columns() → dict[str, sqlalchemy.orm.CompositeProperty][source]: Return a mapping from an attribute name to its CompositeProperty instance for all composite attributes of this class. A composte attribute is an attribute configured using sa.orm.composite.

classmethod hybrid_properties() → dict[str, sqlalchemy.ext.hybrid.hybrid_property][source]: Returns a mapping from an attribute name to a hybrid_property instance. A hybrid_property attribute is a getter style method annotated with the @hybrid_property decorator.

classmethod relationships() → dict[str, sqlalchemy.orm.Relationship][source]: Returns a mapping from an attribute name to a Relationship for all relationships defined on this class.

classmethod table_columns() → list[sqlalchemy.Column][source]: Get a list of the table column names for this model.

classmethod table_insertion_order() → dict[sqlalchemy.Table, int][source]

The insertion priority for each table. A lower number has a higher priority and should be inserted before a table with a higher number. The priorities are determined using a topological sort of the dependency tree created by the relationships between models.

Returns:: A mapping from an SqlAlchemy table to its insertion priority (lower numbers indicate higher priority).
Return type:: dict[Table, int]

classmethod validate_mapper()[source]: Validate that this model is configured correctly.

class rwskit.sqlalchemy.SqlBinaryExpression[source]

Bases: rwskit.config.YamlConfig

A class that represents the basic binary expression for an SQL column.

column: str[source]: The column name.

operator: SqlOperator[source]: The operator to compare the column and value with.

value: Any[source]: The value used as a comparison.

__call__(model_or_table: Type[B] | sqlalchemy.Table) → sqlalchemy.BinaryExpression[source]

to_expression(model_or_table: Type[B] | sqlalchemy.Table) → sqlalchemy.BinaryExpression[source]

Return a clause that can be used with an SqlAlchemy where statement.

Parameters:: model_or_table (sqlalchemy.Table) – The table object that contains the column.
Returns:: The corresponding SqlAlchemy binary expression.
Return type:: BinaryExpression

class rwskit.sqlalchemy.SqlSelectionCriteria[source]

Bases: rwskit.config.YamlConfig

A class that represents a conjunction of SqlBinaryExpression.

expressions: list[SqlBinaryExpression][source]: The list of binary expressions that will be used to filter the query.

to_conjunction(model_or_table: Type[B] | sqlalchemy.Table) → sqlalchemy.BinaryExpression | sqlalchemy.BooleanClauseList[source]

Return a conjunction of binary expressions that can be used with an SqlAlchemy where statement.

Return type:: BinaryExpression | BooleanClauseList

class rwskit.sqlalchemy.SqlOrderExpression[source]

Bases: rwskit.config.YamlConfig

A class that represents a basic order expression for an SQL column.

column: str[source]: The name of the column to sort by.

ascending: bool = True[source]: Whether to sort the column in ascending order.

to_expression(model_or_table: Type[B] | sqlalchemy.Table) → sqlalchemy.UnaryExpression[source]: Convert the order expression to a SqlAlchemy UnaryExpression.

class rwskit.sqlalchemy.SqlOrderCriteria[source]

Bases: rwskit.config.YamlConfig

A class that represents a list of SqlOrderExpressions.

expressions: list[SqlOrderExpression][source]: The list of order expressions.

to_criteria(model_or_table: Type[B] | sqlalchemy.Table) → list[sqlalchemy.UnaryExpression][source]: Convert the order criteria to a list of SqlAlchemy UnaryExpressions that can be used with an SqlAlchemy order_by statement.

class rwskit.sqlalchemy.DatabaseConnectionConfig[source]

Bases: abc.ABC

The options necessary to connect to a database using SqlAlchemy.

database: str[source]: The name of the database or the path to the database file if using sqlite3.

drivername: str = 'sqlite'[source]: The name of the driver used to connect to the database.

username: str | None = None[source]: The user name to use when connecting to the database, if needed.

password: str | None = None[source]: The password to use when connecting to the database, if needed.

host: str | None = None[source]: The database host, if applicable.

port: int | None = None[source]: The database port, if applicable.

use_async: bool = False[source]: Whether to use async mode.

property url: sqlalchemy.URL[source]: Return the sqlalchemy.URL representation of this class.

class rwskit.sqlalchemy.Repository(engine: AlchemyEngine, model_class: Type[M])[source]

A class implementing the basic CRUD operations for the data access layer.

engine[source]

model_class[source]

insert(instances: BaseModel | Iterable[BaseModel], session: sqlalchemy.orm.Session | None = None)[source]

Insert one or more model instances into the database.

Parameters:

instances (BaseModel | Iterable[BaseModel]) – The data to insert.
session (Session | AsyncSession, optional) – A session to use.

Raises:

Exception – If there is a problem adding the data to the database.

async async_insert(instances: BaseModel | Iterable[BaseModel], session: sqlalchemy.ext.asyncio.AsyncSession | None = None)[source]

Insert one or more model instances into the database.

Parameters:

instances (BaseModel | Iterable[BaseModel]) – The data to insert.
session (Session | AsyncSession, optional) – A session to use.

Raises:

Exception – If there is a problem adding the data to the database.

upsert(instances: BaseModel | Iterable[BaseModel], on_conflict: ConflictResolutionStrategy = 'do_nothing', session: sqlalchemy.orm.Session | None = None)[source]

Upsert one or more model instances into the database.

..note::: This only works with dialects that support INSERT … ON CONFLICT. This should include postgresql, mysql/mariadb, and sqlite. However, currently only postgresql is supported.
..note::: This method creates several copies of the data, so you should be careful about memory management if you are inserting a large number of objects.
..warning::: This only works if all non-null fields are provided including primary and foreign keys.
..warning::: This is not well tested with complex model configurations such as hybrid properties and column properties.

Parameters:

instances
on_conflict
session

async async_upsert(instances: BaseModel | Iterable[BaseModel], on_conflict: ConflictResolutionStrategy = 'do_nothing', session: sqlalchemy.ext.asyncio.AsyncSession | None = None)[source]

Upsert one or more model instances into the database.

..note::: This only works with dialects that support INSERT … ON CONFLICT. This should include postgresql, mysql/mariadb, and sqlite. However, currently only postgresql is supported.
..note::: This method creates several copies of the data, so you should be careful about memory management if you are inserting a large number of objects.
..warning::: This only works if all non-null fields are provided including primary and foreign keys.
..warning::: This is not well tested with complex model configurations such as hybrid properties and column properties.

Parameters:

instances
on_conflict
session

find_all(filter_by: Iterable[SqlBinaryExpression] = (), order_by: Iterable[SqlOrderExpression] = (), limit: int | None = None, session: sqlalchemy.orm.Session | None = None) → list[DtoModel][source]

Finds and retrieves multiple models from the database based on specified filters, ordering, and a limit. Converts the retrieved models to their corresponding DTO (Data Transfer Object) representation.

Parameters:

filter_by (Iterable[SqlBinaryExpression], optional) – Collection of SQL binary expressions defining the conditions for filtering the query. Default is an empty iterable.
order_by (Iterable[SqlOrderExpression], optional) – Collection of SQL order expressions defining the sorting order for the query. Default is an empty iterable.
limit (int, optional) – Maximum number of records to retrieve. If None, no limit is applied. Default is None.
session (Session, optional) – An optional database session instance. If not provided, a new session will be created internally for executing the query.

Returns:

A list of DTO instances that correspond to the retrieved database models.

Return type:

list[DtoModel]

async async_find_all(filter_by: Iterable[SqlBinaryExpression] = (), order_by: Iterable[SqlOrderExpression] = (), limit: int | None = None, session: sqlalchemy.orm.Session | None = None) → list[DtoModel][source]

Asynchronously, finds and retrieves multiple models from the database based on specified filters, ordering, and a limit. Converts the retrieved models to their corresponding DTO (Data Transfer Object) representation.

Parameters:

filter_by (Iterable[SqlBinaryExpression], optional) – Collection of SQL binary expressions defining the conditions for filtering the query. Default is an empty iterable.
order_by (Iterable[SqlOrderExpression], optional) – Collection of SQL order expressions defining the sorting order for the query. Default is an empty iterable.
limit (int, optional) – Maximum number of records to retrieve. If None, no limit is applied. Default is None.
session (Session, optional) – An optional database session instance. If not provided, a new session will be created internally for executing the query.

Returns:

A list of DTO instances that correspond to the retrieved database models.

Return type:

list[DtoModel]

find_one(filter_by: Iterable[SqlBinaryExpression] = (), session: sqlalchemy.orm.Session | None = None, raise_on_none: bool = True) → DtoModel[source]

Retrieve a single record from the database that matches the provided filter criteria. Converts the obtained database model into a data transfer object (DTO). An exception is raised when more than one result is returned or no results are found when raise_on_none is True.

Parameters:

filter_by (Iterable[SqlBinaryExpression], optional) – Filter conditions to apply when querying the database. Defaults to an empty tuple.
session (Optional[Session], optional) – SQLAlchemy session to use for querying. If not provided, a new session will be created and used for the operation.
raise_on_none (bool, optional) – Specifies whether to raise an exception if no records match the specified filter criteria. Defaults to True.

Returns:

The retrieved data model in DTO form if a record is found; otherwise, None when raise_on_none is set to False.

Return type:

DtoModel

async async_find_one(filter_by: Iterable[SqlBinaryExpression] = (), session: sqlalchemy.orm.Session | None = None, raise_on_none: bool = True) → DtoModel[source]

Asynchronously, retrieve a single record from the database that matches the provided filter criteria. Converts the obtained database model into a data transfer object (DTO). An exception is raised when more than one result is returned or no results are found when raise_on_none is True.

Parameters:

filter_by (Iterable[SqlBinaryExpression], optional) – Filter conditions to apply when querying the database. Defaults to an empty tuple.
session (Optional[Session], optional) – SQLAlchemy session to use for querying. If not provided, a new session will be created and used for the operation.
raise_on_none (bool, optional) – Specifies whether to raise an exception if no records match the specified filter criteria. Defaults to True.

Returns:

The retrieved data model in DTO form if a record is found; otherwise, None when raise_on_none is set to False.

Return type:

DtoModel