rwskit.sqlalchemy ================= .. py:module:: rwskit.sqlalchemy .. autoapi-nested-parse:: Utilities for working with SqlAlchemy. Attributes ---------- .. autoapisummary:: rwskit.sqlalchemy.log rwskit.sqlalchemy.B rwskit.sqlalchemy.M rwskit.sqlalchemy.TableArgs rwskit.sqlalchemy.DtoModel rwskit.sqlalchemy.SqlOperator rwskit.sqlalchemy.ConflictResolutionStrategy rwskit.sqlalchemy.Walkable rwskit.sqlalchemy.WalkPredicate Classes ------- .. autoapisummary:: rwskit.sqlalchemy.AlchemyEngine rwskit.sqlalchemy.BaseModel rwskit.sqlalchemy.SqlBinaryExpression rwskit.sqlalchemy.SqlSelectionCriteria rwskit.sqlalchemy.SqlOrderExpression rwskit.sqlalchemy.SqlOrderCriteria rwskit.sqlalchemy.DatabaseConnectionConfig rwskit.sqlalchemy.Repository Functions --------- .. autoapisummary:: rwskit.sqlalchemy.handle_after_mapper_configured Module Contents --------------- .. py:data:: log .. py:data:: B A type extending :class:`~sqlalchemy.orm.DeclarativeBase`. .. py:data:: M .. py:data:: TableArgs The type of the ``__table_args__`` attribute on a :class:`~sqlalchemy.Table`. .. py:data:: DtoModel The base type for a DTO model. .. py:data:: SqlOperator The supported SQL operators for use in an :class:`SqlBinaryExpression`. .. py:data:: ConflictResolutionStrategy .. py:data:: Walkable .. py:data:: WalkPredicate .. py:class:: AlchemyEngine(engine_or_config: sqlalchemy.ext.asyncio.AsyncEngine | sqlalchemy.Engine | DatabaseConnectionConfig, base_model: Type[sqlalchemy.orm.DeclarativeBase], **engine_kwargs: Any) A wrapper around :class:`sqlalchemy.engine.Engine` that provides additional functionality. .. py:attribute:: base_model .. py:property:: dialect :type: str Returns the dialect name of the engine. .. py:property:: supports_async :type: bool Returns true if the engine supports async mode. .. py:property:: sync_engine :type: sqlalchemy.Engine Get a reference to a synchronous :class:`sqlalchemy.engine.Engine`. .. py:property:: async_engine :type: sqlalchemy.ext.asyncio.AsyncEngine Get a reference to an asynchronous :class:`sqlalchemy.ext.asyncio.AsyncEngine`. :returns: The engine. :rtype: AsyncEngine :raises ValueError: If the engine does not support async mode. .. py:method:: make_session() -> sqlalchemy.Session Create a session from the :attr:`AlchemyEngine.session_factory`. ..note:: You probably want to use :meth:`AlchemyEngine.session_scope`, but there may be cases where this is more appropriate. :returns: A :class:`~sqlalchemy.orm.Session` object. :rtype: Session .. py:method:: make_connection() -> sqlalchemy.Connection Get a connection to the engine. :returns: An :class:`sqlalchemy.Connection`. :rtype: Connection .. py:method:: make_raw_connection() -> sqlalchemy.PoolProxiedConnection 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 :class:`sqlalchemy.PoolProxiedConnection`. :rtype: PoolProxiedConnection .. py:method:: session_scope(expire_on_commit: bool = False) -> sqlalchemy.Session A context manager for committing successful transactions when the session is complete or rolling back if there was an exception. :param expire_on_commit: 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. :type expire_on_commit: bool, default=False :returns: An :class:`sqlalchemy.orm.Session`. :rtype: Session .. py:method:: async_session_scope(expire_on_commit: bool = False) -> sqlalchemy.Session :async: A context manager for committing successful transactions when the session is complete or rolling back if there was an exception. :param expire_on_commit: 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. :type expire_on_commit: bool, default=False :returns: An :class:`sqlalchemy.orm.Session`. :rtype: Session .. py:method:: test_scope(expire_on_commit: bool = False) -> sqlalchemy.Session 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. :param expire_on_commit: 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. :type expire_on_commit: bool, default=False :returns: An :class:`sqlalchemy.orm.Session`. :rtype: Session :raises RuntimeError: If the user tries to commit changes during the session. .. py:method:: async_test_scope(expire_on_commit: bool = False) -> sqlalchemy.Session :async: 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. :param expire_on_commit: 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. :type expire_on_commit: bool, default=False :returns: An :class:`sqlalchemy.orm.Session`. :rtype: Session :raises RuntimeError: If the user tries to commit changes during the session. .. py:function:: handle_after_mapper_configured(mapper: sqlalchemy.orm.Mapper, cls: Type) Validate the configuration and create an index on the natural key. .. py:class:: BaseModel Bases: :py:obj:`sqlalchemy.orm.DeclarativeBase`, :py:obj:`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 :meth:`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 :meth:`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 :meth:`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 :meth:`from_dict` method does not handle cyclic relationships and will typically not be able to associate a parent instance from the child. .. py:attribute:: __abstract__ :value: True .. py:attribute:: metadata :type: sqlalchemy.MetaData .. py:attribute:: hasher The hasher used to calculate the has value of objects. .. py:method:: __hash__() -> int .. py:method:: __eq__(other: Any) -> bool .. py:method:: __tablename__() .. py:method:: merge_table_args(new_args: TableArgs = ()) -> TableArgs :classmethod: 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. :param new_args: Additional table arguments to be merged with the arguments of our ancestors. :type new_args: TableArgs :returns: The merged table arguments as a tuple. The first ``N`` arguments of the tuple contain positional arguments passed to the constructor of :class:`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. :rtype: TableArgs .. rubric:: 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] .. py:method:: find_by_table_name(table_name: str) -> Optional[Type[BaseModel]] :classmethod: Find a model derived from this class by its table name. :param table_name: The name of the table whose model class you want to find. :type table_name: str :returns: Returns the model class if the table is found, otherwise ``None``. :rtype: Type[FindByNameBase], optional .. py:property:: primary_key :type: tuple[tuple[str, Any], Ellipsis] 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. .. py:property:: natural_key :type: tuple[tuple[str, Optional[Any]], Ellipsis] 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 :meth:`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. .. py:method:: as_table_dict() -> dict[str, Any] 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. :rtype: dict[str, Any] .. py:method:: to_dict(exclude_attributes: Iterable[str] = (), drop_none: bool = False) -> dict[str, Any] 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 .. py:method:: from_dict(data: dict) -> BaseModel :classmethod: 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. :param data: 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). :type data: dict :returns: An instance of the BaseModel subclass created from the dictionary input. :rtype: BaseModel .. py:method:: dto_module() -> types.ModuleType :classmethod: Get the module where the DTO class is defined. .. py:method:: dto_module_name() -> str :classmethod: Get the name of the module where the DTO class is defined. .. py:method:: dto_class_name() -> str :classmethod: Get the name of the DTO class. .. py:method:: dto_import_path() -> str :classmethod: Get the import path of the DTO class. .. py:method:: dto_exclude_attributes() -> set[str] :classmethod: A set of attribute names to exclude from the DTO representation. .. py:method:: to_dto_class() -> Type[pydantic.BaseModel] :classmethod: 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. :rtype: Type[pydantic.BaseModel] .. py:method:: to_dto() -> pydantic.BaseModel Convert this model to a DTO instance. :returns: The equivalent DTO instance. :rtype: pydantic.BaseModel :raises pydantic.ValidationError: If any of the required fields are not present. **Note**: this includes database generated fields, such as ``ids``. .. py:method:: from_dto(dto: pydantic.BaseModel) -> M :classmethod: Create a model instance from a DTO instance. .. py:method:: walk_children(callback: Callable[[B], None], traverse_viewonly: bool = True) A method to traverse the relationships of a given instance and apply a callback to each node in the traversal. :param callback: The function to call on each traversed node. :type callback: Callable[[DeclarativeBase], None] :param traverse_viewonly: Whether to traverse viewonly relationships. :type traverse_viewonly: bool, default=True .. py:method:: copy() -> M 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. :rtype: BaseModel .. py:method:: class_name() -> str :classmethod: Get the name of the class including the package and module prefix. .. py:method:: fields() -> dict[str, dataclasses.Field] :classmethod: Return a mapping of the field names to their corresponding ``Field`` objects. Note, the dictionary is ordered by the field name. .. py:method:: primary_key_columns() -> dict[str, sqlalchemy.orm.ColumnProperty] :classmethod: Return the list of primary key columns of this class. .. py:method:: natural_key_columns() -> dict[str, sqlalchemy.orm.ColumnProperty] :classmethod: Return the list of natural key columns of this class. .. py:method:: regular_columns() -> dict[str, sqlalchemy.orm.ColumnProperty] :classmethod: 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`. .. py:method:: column_properties() -> dict[str, sqlalchemy.orm.ColumnProperty] :classmethod: 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``. .. py:method:: composite_columns() -> dict[str, sqlalchemy.orm.CompositeProperty] :classmethod: 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``. .. py:method:: hybrid_properties() -> dict[str, sqlalchemy.ext.hybrid.hybrid_property] :classmethod: 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. .. py:method:: relationships() -> dict[str, sqlalchemy.orm.Relationship] :classmethod: Returns a mapping from an attribute name to a ``Relationship`` for all relationships defined on this class. .. py:method:: table_columns() -> list[sqlalchemy.Column] :classmethod: Get a list of the table column names for this model. .. py:method:: table_insertion_order() -> dict[sqlalchemy.Table, int] :classmethod: 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). :rtype: dict[Table, int] .. py:method:: validate_mapper() :classmethod: Validate that this model is configured correctly. .. py:class:: SqlBinaryExpression Bases: :py:obj:`rwskit.config.YamlConfig` A class that represents the basic binary expression for an SQL column. .. py:attribute:: column :type: str The column name. .. py:attribute:: operator :type: SqlOperator The operator to compare the ``column`` and ``value`` with. .. py:attribute:: value :type: Any The value used as a comparison. .. py:method:: __call__(model_or_table: Type[B] | sqlalchemy.Table) -> sqlalchemy.BinaryExpression .. py:method:: to_expression(model_or_table: Type[B] | sqlalchemy.Table) -> sqlalchemy.BinaryExpression Return a clause that can be used with an SqlAlchemy ``where`` statement. :param model_or_table: The table object that contains the column. :type model_or_table: sqlalchemy.Table :returns: The corresponding SqlAlchemy binary expression. :rtype: BinaryExpression .. py:class:: SqlSelectionCriteria Bases: :py:obj:`rwskit.config.YamlConfig` A class that represents a conjunction of SqlBinaryExpression. .. py:attribute:: expressions :type: list[SqlBinaryExpression] The list of binary expressions that will be used to filter the query. .. py:method:: to_conjunction(model_or_table: Type[B] | sqlalchemy.Table) -> sqlalchemy.BinaryExpression | sqlalchemy.BooleanClauseList Return a conjunction of binary expressions that can be used with an SqlAlchemy ``where`` statement. :rtype: BinaryExpression | BooleanClauseList .. py:class:: SqlOrderExpression Bases: :py:obj:`rwskit.config.YamlConfig` A class that represents a basic order expression for an SQL column. .. py:attribute:: column :type: str The name of the column to sort by. .. py:attribute:: ascending :type: bool :value: True Whether to sort the column in ascending order. .. py:method:: to_expression(model_or_table: Type[B] | sqlalchemy.Table) -> sqlalchemy.UnaryExpression Convert the order expression to a SqlAlchemy ``UnaryExpression``. .. py:class:: SqlOrderCriteria Bases: :py:obj:`rwskit.config.YamlConfig` A class that represents a list of ``SqlOrderExpressions``. .. py:attribute:: expressions :type: list[SqlOrderExpression] The list of order expressions. .. py:method:: to_criteria(model_or_table: Type[B] | sqlalchemy.Table) -> list[sqlalchemy.UnaryExpression] Convert the order criteria to a list of SqlAlchemy ``UnaryExpressions`` that can be used with an SqlAlchemy ``order_by`` statement. .. py:class:: DatabaseConnectionConfig Bases: :py:obj:`abc.ABC` The options necessary to connect to a database using SqlAlchemy. .. py:attribute:: database :type: str The name of the database or the path to the database file if using sqlite3. .. py:attribute:: drivername :type: str :value: 'sqlite' The name of the driver used to connect to the database. .. py:attribute:: username :type: Optional[str] :value: None The user name to use when connecting to the database, if needed. .. py:attribute:: password :type: Optional[str] :value: None The password to use when connecting to the database, if needed. .. py:attribute:: host :type: Optional[str] :value: None The database host, if applicable. .. py:attribute:: port :type: Optional[int] :value: None The database port, if applicable. .. py:attribute:: use_async :type: bool :value: False Whether to use async mode. .. py:property:: url :type: sqlalchemy.URL Return the :class:`sqlalchemy.URL` representation of this class. .. py:class:: Repository(engine: AlchemyEngine, model_class: Type[M]) A class implementing the basic CRUD operations for the data access layer. .. py:attribute:: engine .. py:attribute:: model_class .. py:method:: insert(instances: BaseModel | Iterable[BaseModel], session: Optional[sqlalchemy.orm.Session] = None) Insert one or more model instances into the database. :param instances: The data to insert. :type instances: BaseModel | Iterable[BaseModel] :param session: A session to use. :type session: Session | AsyncSession, optional :raises Exception: If there is a problem adding the data to the database. .. py:method:: async_insert(instances: BaseModel | Iterable[BaseModel], session: Optional[sqlalchemy.ext.asyncio.AsyncSession] = None) :async: Insert one or more model instances into the database. :param instances: The data to insert. :type instances: BaseModel | Iterable[BaseModel] :param session: A session to use. :type session: Session | AsyncSession, optional :raises Exception: If there is a problem adding the data to the database. .. py:method:: upsert(instances: BaseModel | Iterable[BaseModel], on_conflict: ConflictResolutionStrategy = 'do_nothing', session: Optional[sqlalchemy.orm.Session] = None) 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. :param instances: :param on_conflict: :param session: .. py:method:: async_upsert(instances: BaseModel | Iterable[BaseModel], on_conflict: ConflictResolutionStrategy = 'do_nothing', session: Optional[sqlalchemy.ext.asyncio.AsyncSession] = None) :async: 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. :param instances: :param on_conflict: :param session: .. py:method:: find_all(filter_by: Iterable[SqlBinaryExpression] = (), order_by: Iterable[SqlOrderExpression] = (), limit: Optional[int] = None, session: Optional[sqlalchemy.orm.Session] = None) -> list[DtoModel] 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. :param filter_by: Collection of SQL binary expressions defining the conditions for filtering the query. Default is an empty iterable. :type filter_by: Iterable[SqlBinaryExpression], optional :param order_by: Collection of SQL order expressions defining the sorting order for the query. Default is an empty iterable. :type order_by: Iterable[SqlOrderExpression], optional :param limit: Maximum number of records to retrieve. If None, no limit is applied. Default is None. :type limit: int, optional :param session: An optional database session instance. If not provided, a new session will be created internally for executing the query. :type session: Session, optional :returns: A list of DTO instances that correspond to the retrieved database models. :rtype: list[DtoModel] .. py:method:: async_find_all(filter_by: Iterable[SqlBinaryExpression] = (), order_by: Iterable[SqlOrderExpression] = (), limit: Optional[int] = None, session: Optional[sqlalchemy.orm.Session] = None) -> list[DtoModel] :async: 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. :param filter_by: Collection of SQL binary expressions defining the conditions for filtering the query. Default is an empty iterable. :type filter_by: Iterable[SqlBinaryExpression], optional :param order_by: Collection of SQL order expressions defining the sorting order for the query. Default is an empty iterable. :type order_by: Iterable[SqlOrderExpression], optional :param limit: Maximum number of records to retrieve. If None, no limit is applied. Default is None. :type limit: int, optional :param session: An optional database session instance. If not provided, a new session will be created internally for executing the query. :type session: Session, optional :returns: A list of DTO instances that correspond to the retrieved database models. :rtype: list[DtoModel] .. py:method:: find_one(filter_by: Iterable[SqlBinaryExpression] = (), session: Optional[sqlalchemy.orm.Session] = None, raise_on_none: bool = True) -> DtoModel 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. :param filter_by: Filter conditions to apply when querying the database. Defaults to an empty tuple. :type filter_by: Iterable[SqlBinaryExpression], optional :param session: SQLAlchemy session to use for querying. If not provided, a new session will be created and used for the operation. :type session: Optional[Session], optional :param raise_on_none: Specifies whether to raise an exception if no records match the specified filter criteria. Defaults to True. :type raise_on_none: bool, optional :returns: The retrieved data model in DTO form if a record is found; otherwise, None when `raise_on_none` is set to False. :rtype: DtoModel .. py:method:: async_find_one(filter_by: Iterable[SqlBinaryExpression] = (), session: Optional[sqlalchemy.orm.Session] = None, raise_on_none: bool = True) -> DtoModel :async: 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. :param filter_by: Filter conditions to apply when querying the database. Defaults to an empty tuple. :type filter_by: Iterable[SqlBinaryExpression], optional :param session: SQLAlchemy session to use for querying. If not provided, a new session will be created and used for the operation. :type session: Optional[Session], optional :param raise_on_none: Specifies whether to raise an exception if no records match the specified filter criteria. Defaults to True. :type raise_on_none: bool, optional :returns: The retrieved data model in DTO form if a record is found; otherwise, None when `raise_on_none` is set to False. :rtype: DtoModel