Skip to content

API Reference

This is the API reference for the ActiveRecordMixin class.

Table of Contents

Instance Methods

fill

def fill(**kwargs)

Fills the object with values from kwargs without saving to the database.

Parameters:

  • kwargs: Key-value pairs of attributes to set.

Returns:

  • Self: The instance itself for method chaining.

Raises:

  • KeyError: If attribute doesn't exist.

Example:

user.fill(name='Bob', age=30)

save

async def save()

Saves the current row to the database.

Returns:

  • Self: The saved instance for method chaining.

Raises: Any database errors are caught and will trigger a rollback.

Example:

user = User(name='Bob')
await user.save()

update

async def update(**kwargs)

Updates the current row with the provided values.

Parameters:

  • kwargs: Key-value pairs of attributes to update.

Returns:

  • Self: The updated instance for method chaining.

Raises: Any database errors are caught and will trigger a rollback.

Example:

await user.update(name='Bob2', age=31)

edit

async def edit(**kwargs)

Synonym for update().

delete

async def delete()

Deletes the current row from the database.

Raises: Any database errors are caught and will trigger a rollback.

Example:

await user.delete()

remove

async def remove()

Synonym for delete().

Class Methods

create

async def create(**kwargs)

Creates a new row with the provided values.

Parameters:

  • kwargs: Key-value pairs for the new instance.

Returns:

  • Self: The created instance for method chaining.

Raises: Any database errors are caught and will trigger a rollback.

Example:

user = await User.create(name='Bob', age=30)

insert

async def insert(**kwargs)

Synonym for create().

add

async def add(**kwargs)

Synonym for create().

save_all

async def save_all(rows: Sequence[Self], refresh: bool = False)

Saves multiple rows in a single transaction.

Parameters:

  • rows: Sequence of model instances to save.
  • refresh: Whether to refresh the instances after saving (default: False).

Raises: Any database errors are caught and will trigger a rollback.

Example:

users = [User(name='Bob'), User(name='Alice')]
await User.save_all(users)

create_all

async def create_all(rows: Sequence[Self], refresh: bool = False)

Synonym for save_all() when creating new rows.

update_all

async def update_all(rows: Sequence[Self], refresh: bool = False)

Synonym for save_all() when updating existing rows.

delete_all

async def delete_all(rows: Sequence[Self])

Deletes multiple rows in a single transaction.

Parameters:

  • rows: Sequence of model instances to delete.

Raises: Any database errors are caught and will trigger a rollback.

Example:

users = await User.where(age__lt=18).all()
await User.delete_all(users)

destroy

async def destroy(*ids: object)

Deletes multiple rows by their primary keys.

Parameters:

  • ids: Primary key values of rows to delete.

Raises: Any database errors are caught and will trigger a rollback.

Example:

await User.destroy(1, 2, 3)  # Deletes users with IDs 1, 2, and 3

get

async def get(
    pk: object,
    join: list[QueryableAttribute | tuple[QueryableAttribute, bool]] | None = None,
    subquery: list[QueryableAttribute | tuple[QueryableAttribute, bool]] | None = None,
    schema: dict[InstrumentedAttribute, str | tuple[str, dict[InstrumentedAttribute, Any]] | dict] | None = None,
)

Fetches a row by primary key.

Parameters:

  • pk: Primary key value.
  • join: Paths to join eager load. See the docs of join method for details.
  • subquery: Paths to subquery eager load. See the docs of with_subquery method for details.
  • schema: Schema for the eager loading. See the docs of with_schema method for details.

Returns:

  • Self | None: Instance for method chaining or None if not found.

Raises:

  • MultipleResultsFound: If multiple rows match.

Example:

user = await User.get(1)

get_or_fail

async def get_or_fail(
    pk: object,
    join: list[QueryableAttribute | tuple[QueryableAttribute, bool]] | None = None,
    subquery: list[QueryableAttribute | tuple[QueryableAttribute, bool]] | None = None,
    schema: dict[InstrumentedAttribute, str | tuple[str, dict[InstrumentedAttribute, Any]] | dict] | None = None,
)

Fetches a row by primary key or raises an exception if not found.

Parameters:

  • pk: Primary key value.
  • join: Paths to join eager load. See the docs of join method for details.
  • subquery: Paths to subquery eager load. See the docs of with_subquery method for details.
  • schema: Schema for the eager loading. See the docs of with_schema method for details.

Returns:

  • Self: Instance for method chaining.

Raises:

  • NoResultFound: If no row is found.
  • MultipleResultsFound: If multiple rows match.

Example:

user = await User.get_or_fail(1)  # Raises if not found

options

def options(*args: ExecutableOption)

Creates a query and applies the given list of mapper options.

Warning

Quoting from SQLAlchemy docs:

When including `joinedload()` in reference to a one-to-many or
many-to-many collection, the `Result.unique()` method must be
applied to the returned result, which will make the incoming rows
unique by primary key that otherwise are multiplied out by the join.
The ORM will raise an error if this is not present.

This is not automatic in modern SQLAlchemy, as it changes the behavior
of the result set to return fewer ORM objects than the statement would
normally return in terms of number of rows. Therefore SQLAlchemy keeps
the use of Result.unique() explicit, so there is no ambiguity that the
returned objects are made unique on primary key.

To learn more about options, see sqlalchemy.orm.Query.options docs.

Parameters:

  • args: Mapper options.

Returns:

Example:

users = await User.options(joinedload(User.posts)).unique_all()

user = await User.options(joinedload(User.posts)).first()

users = await User.options(subqueryload(User.posts)).all()

filter

def filter(*criterion: _ColumnExpressionArgument[bool], **filters: Any)

Creates a filtered query using SQLAlchemy or Django-style filters.

Parameters:

  • criterion: SQLAlchemy style filter expressions.
  • filters: Django-style filters.

Returns:

Example:

# SQLAlchemy style
users = await User.filter(User.age >= 18).all()

# Django style
users = await User.filter(age__gte=18).all()

# Mixed
users = await User.filter(User.age >= 18, name__like='%Bob%').all()

where

def where(*criterion: _ColumnExpressionArgument[bool], **filters: Any)

Synonym for filter().

find

def find(*criterion: _ColumnExpressionArgument[bool], **filters: Any)

Synonym for filter().

find_one

async def find_one(*criterion: _ColumnExpressionArgument[bool], **filters: Any)

Finds a single row matching the criteria.

This is same as calling await cls.find(*criterion, **filters).one().

Returns:

  • Self: Instance for method chaining.

Raises:

  • NoResultFound: If no row is found.
  • MultipleResultsFound: If multiple rows match.

Example:

user = await User.find_one(name='Bob')  # Raises if not found

find_one_or_none

async def find_one_or_none(*criterion: _ColumnExpressionArgument[bool], **filters: Any)

Finds a single row matching the criteria or None.

This is same as calling await cls.find(*criterion, **filters).one_or_none().

Returns:

  • Self | None: Instance for method chaining or None.

Raises:

  • MultipleResultsFound: If multiple rows match.

Example:

user = await User.find_one_or_none(name='Bob')  # Returns None if not found

find_all

async def find_all(*criterion: _ColumnExpressionArgument[bool], **filters: Any)

Finds all rows matching the criteria.

This is same as calling await cls.find(*criterion, **filters).all().

Returns:

  • list[Self]: List of instances for method chaining.

Example:

users = await User.find_all(age__gte=18)

find_first

async def find_first(*criterion: _ColumnExpressionArgument[bool], **filters: Any)

Finds a single row matching the criteria or None.

This is same as calling await cls.find(*criterion, **filters).first().

Returns:

  • Self | None: Instance for method chaining or None.

Example:

user = await User.find_first(name='Bob')

find_unique

async def find_unique(*criterion: _ColumnExpressionArgument[bool], **filters: Any)

Finds all unique rows matching the criteria and returns an ScalarResult object with them.

This is same as calling await cls.find(*criterion, **filters).unique().

Returns:

  • sqlalchemy.engine.ScalarResult: Scalars.

Example:

users_scalars = await User.find_unique(name__like='%John%')
users = users_scalars.all()

find_unique_all

async def find_unique_all(*criterion: _ColumnExpressionArgument[bool], **filters: Any)

Finds all unique rows matching the criteria and returns a list.

This is same as calling await cls.find(*criterion, **filters).unique_all().

Returns:

  • list[Self]: List of instances.

Example:

users = await User.find_unique_all(name__like='%John%')

find_unique_first

async def find_unique_first(*criterion: _ColumnExpressionArgument[bool], **filters: Any)

Finds a single unique row matching the criteria or None.

This is same as calling await cls.find(*criterion, **filters).unique_first().

Returns:

  • Self | None: Instance for method chaining or None.

Example:

user = await User.find_unique_first(name__like='%John%', age=30)

find_unique_one

async def find_unique_one(*criterion: _ColumnExpressionArgument[bool], **filters: Any)

Finds a single unique row matching the criteria.

This is same as calling await cls.find(*criterion, **filters).unique_one().

Returns:

  • Self: Instance for method chaining.

Raises:

  • NoResultFound: If no row is found.
  • MultipleResultsFound: If multiple rows match.

Example:

user = await User.find_unique_one(name__like='%John%', age=30)

find_unique_one_or_none

async def find_unique_one_or_none(*criterion: _ColumnExpressionArgument[bool], **filters: Any)

Finds a single unique row matching the criteria or None.

This is same as calling await cls.find(*criterion, **filters).unique_one_or_none().

Returns:

  • Self | None: Instance for method chaining or None.

Raises:

  • MultipleResultsFound: If multiple rows match.

Example:

user = await User.find_unique_one_or_none(name__like='%John%', age=30)

order_by

def order_by(*columns: _ColumnExpressionOrStrLabelArgument[Any])

Creates a query with ORDER BY clause.

Parameters:

  • columns: Column names or SQLAlchemy column expressions.

Returns:

Example:

# String column names (Django style)
users = await User.order_by('-created_at', 'name').all()

# SQLAlchemy expressions
users = await User.order_by(User.created_at.desc(), User.name).all()

sort

def sort(*columns: _ColumnExpressionOrStrLabelArgument[Any])

Synonym for order_by().

offset

def offset(offset: int)

Creates a query with OFFSET clause.

Parameters:

  • offset: Number of rows to skip.

Returns:

Raises:

  • ValueError: If offset is negative.

Example:

users = await User.offset(10).all()

skip

def skip(skip: int)

Synonym for offset().

limit

def limit(limit: int)

Creates a query with LIMIT clause.

Parameters:

  • limit: Maximum number of rows to return.

Returns:

Raises:

  • ValueError: If limit is negative.

Example:

users = await User.limit(5).all()

take

def take(take: int)

Synonym for limit().

join

def join(*paths: QueryableAttribute | tuple[QueryableAttribute, bool])

Creates a query with LEFT OUTER JOIN eager loading.

When a tuple is passed, the second element must be boolean. If it is True, the join is INNER JOIN, otherwise LEFT OUTER JOIN.

Parameters:

  • paths: Relationship attributes to join.

Returns:

Example:

comments = await Comment.join(
    Comment.user,
    (Comment.post, True)  # True means INNER JOIN
).all()

with_subquery

def with_subquery(*paths: QueryableAttribute | tuple[QueryableAttribute, bool])

Creates a query with subquery or selectin loading.

Emits a second SELECT statement (Subqueryload) for each relationship to be loaded, across all result objects at once.

When a tuple is passed, the second element must be boolean. If it is True, the eager loading strategy is SELECT IN (Selectinload), otherwise SELECT JOIN (Subqueryload).

Warning

A query which makes use of subqueryload() in conjunction with a limiting modifier such as Query.limit() or Query.offset() should always include Query.order_by() against unique column(s) such as the primary key, so that the additional queries emitted by subqueryload() include the same ordering as used by the parent query. Without it, there is a chance that the inner query could return the wrong rows, as specified in SQLAlchemy docs.

# incorrect, no ORDER BY
User.options(subqueryload(User.addresses)).first()
# incorrect if User.name is not unique
User.options(subqueryload(User.addresses)).order_by(User.name).first()
# correct
User.options(subqueryload(User.addresses)).order_by(
    User.name, User.id
).first()

Parameters:

  • paths: Relationship attributes to load.

Returns:

Example:

users = await User.with_subquery(
    User.posts,
    (User.comments, True)  # True means selectin loading
).all()

with_schema

def with_schema(
    schema: dict[InstrumentedAttribute, str | tuple[str, dict[InstrumentedAttribute, Any]] | dict]
)

Creates a query with complex eager loading schema.

Useful for complex cases where you need to load nested relationships in separate queries.

Parameters:

  • schema: Dictionary defining the loading strategy.

Returns:

from sqlactive import JOINED, SUBQUERY
schema = {
    User.posts: JOINED,
    User.comments: (SUBQUERY, {
        Comment.user: JOINED
    })
}
users = await User.with_schema(schema).all()

scalars

async def scalars()

Returns a sqlalchemy.engine.ScalarResult object containing all rows.

Returns:

  • sqlalchemy.engine.ScalarResult: Scalars.

Example:

result = await User.scalars()
users = result.all()

first

async def first()

Fetches the first row.

Returns:

  • Self | None: Instance for method chaining or None if no matches.

Example:

user = await User.first()

one

async def one()

Fetches exactly one row.

Returns:

  • Self: Instance for method chaining.

Raises:

  • NoResultFound: If no row is found.
  • MultipleResultsFound: If multiple rows match.

Example:

user = await User.one()  # Raises if not exactly one match

one_or_none

async def one_or_none()

Fetches exactly one row or None.

Returns:

  • Self | None: Instance for method chaining or None.

Raises:

  • MultipleResultsFound: If multiple rows match.

Example:

user = await User.one_or_none()

fetch_one

async def fetch_one()

Synonym for one().

fetch_one_or_none

async def fetch_one_or_none()

Synonym for one_or_none().

all

async def all()

Fetches all rows.

Returns:

  • list[Self]: List of instances.

Example:

users = await User.all()

fetch_all

async def fetch_all()

Synonym for all().

to_list

async def to_list()

Synonym for all().

unique

async def unique()

Returns a sqlalchemy.engine.ScalarResult object containing all unique rows.

Returns:

  • sqlalchemy.engine.ScalarResult: Scalars.

Example:

result = await User.unique()
users = result.all()

unique_all

async def unique_all()

Fetches all unique rows.

Returns:

  • list[Self]: List of instances.

Example:

users = await User.unique_all()

unique_first

async def unique_first()

Fetches the first unique row.

Returns:

  • Self | None: Instance for method chaining or None.

Example:

user = await User.unique_first()

unique_one

async def unique_one()

Fetches exactly one unique row.

Returns:

  • Self: Instance for method chaining.

Raises:

  • NoResultFound: If no row is found.
  • MultipleResultsFound: If multiple rows match.

Example:

user = await User.unique_one()

unique_one_or_none

async def unique_one_or_none()

Fetches exactly one unique row or None.

Returns:

  • Self | None: Instance for method chaining or None.

Raises:

  • MultipleResultsFound: If multiple rows match.

Example:

user = await User.unique_one_or_none()

smart_query

def smart_query(
    criterion: Sequence[_ColumnExpressionArgument[bool]] | None = None,
    filters: dict[str, Any] | dict[OperatorType, Any] | list[dict[str, Any]] | list[dict[OperatorType, Any]] | None = None,
    sort_columns: Sequence[_ColumnExpressionOrStrLabelArgument[Any]] | None = None,
    sort_attrs: Sequence[str] | None = None,
    schema: dict[InstrumentedAttribute, str | tuple[str, dict[InstrumentedAttribute, Any]] | dict] | None = None,
)

Creates a query combining filtering, sorting, and eager loading.

Parameters:

  • criterion: SQLAlchemy filter expressions.
  • filters: Django-style filters.
  • sort_columns: SQLAlchemy columns to sort by.
  • sort_attrs: String column names to sort by.
  • schema: Eager loading schema.

Returns:

Example:

users = await User.smart_query(
    criterion=(User.age >= 18,),
    filters={'name__like': '%Bob%'},
    sort_columns=(User.username,),
    sort_attrs=['-created_at'],
    schema={User.posts: 'joined'}
).all()