Skip to content

SQL API

SQLEntity

Bases: SQLModel, Schema, HasOrm[Mapped[Any]], BaseSQLEntity[IDT], Generic[IDT, SourceT]

SQL entity with an optional signpost type checked on declaration.

__table__ class-attribute 

__table__: Table

id instance-attribute 

id: ReadField[IDT]

__sql_entity__ classmethod 

__sql_entity__() -> type[Self]

__sql_source__ classmethod 

__sql_source__() -> type[SourceT]

Expose the source type for inference without constructing a source.

__init_subclass__ 

__init_subclass__() -> None

SQLEntityStream

Bases: SimpleAsyncEntityStream[E], AsyncEntityStreamable[E], SQLDataBackend

Entity stream backed by a SQLAlchemy Select.

Operations that SQL can express are pushed down into the statement (take → LIMIT, drop → OFFSET, sort by field → ORDER BY, distinct, count, first, column projection via to_values). Everything else (Python predicates/mappers) falls back to the in-memory stream over a server-side cursor — rows are never fully buffered.

query instance-attribute 

query: BaseQuery[E, Select[tuple[E]]] | None = None

statement instance-attribute 

statement = stmt

__init__ 

__init__(
    stmt: Select[tuple[E]] | BaseQuery[E, Select[tuple[E]]],
    signpost: SQLSignpost
    | Signpostable[AbstractAsyncContextManager[AsyncSession]],
) -> None

clone 

clone(stmt: Select[tuple[E]]) -> SQLEntityStream[E]

__aiter__ 

__aiter__() -> AsyncIterator[E]

join 

join[F: Entitieable, Key: Hashable, LeftOrm, RightOrm](
    other: AsyncEntityStreamable[F],
    *,
    on: tuple[FieldRef[E, Key, LeftOrm], FieldRef[F, Key, RightOrm]],
) -> JoinedStream[E, F]

take 

take(n: int) -> SimpleAsyncEntityStream[E]

drop 

drop(n: int) -> SimpleAsyncEntityStream[E]

distinct 

distinct() -> SimpleAsyncEntityStream[E]

filter 

filter(predicate: PredicateCallable[E]) -> SimpleAsyncEntityStream[E]

sort 

sort[R: RichComparisonable[Any]](
    key: GetterCallable[E, R], reverse: bool = False
) -> AsyncEntityStreamable[E]

count async 

count() -> int

first async 

first() -> E | None

to_values 

to_values[K: HashableAndValuable](
    mapper: MapperCallable[E, K],
) -> AsyncValueStreamable[K]

SQLQuery

Bases: BaseQuery[E, Select[tuple[E]]], Generic[E, SourceT]

Infer the entity's source; source= additionally pins its instance.

source class-attribute instance-attribute 

source: SourceT | None = source

native_conditions property 

native_conditions: bool

__init__ 

__init__(
    entity: SQLEntityClassable[E, SourceT],
    statement: Select[tuple[E]] | None = None,
    limit_value: int | UndefinedType | None = Undefined,
    offset_value: Annotated[int, Ge(0)] = 0,
    conditions: tuple[Expression[E], ...] = (),
    orderings: tuple[tuple[str, bool], ...] = (),
    source: SourceT | None = None,
) -> None

build 

build() -> Select[tuple[E]]

join 

join[F: SQLEntity[Any], L: SQLEntity[Any], Key: Hashable, LO, RO](
    entity: type[SQLEntity[Any, SourceT]],
    condition: FieldComparison[L, F, Key, LO, RO],
) -> SQLJoinedQuery[SourceT, FieldPredicate[F], E, F]

SQLValueStream

Bases: SimpleAsyncValueStream[T], AsyncValueStreamable[T], SQLDataBackend

Single-column stream backed by a SQLAlchemy Select.

Pushes down take/drop/distinct/sort and the aggregate terminals (count, sum, min, max) into SQL; Python callables fall back to the in-memory stream over a server-side cursor.

statement instance-attribute 

statement = stmt

__init__ 

__init__(
    stmt: Select[tuple[T]],
    signpost: SQLSignpost
    | Signpostable[AbstractAsyncContextManager[AsyncSession]],
) -> None

clone 

clone(stmt: Select[tuple[T]]) -> SQLValueStream[T]

__aiter__ 

__aiter__() -> AsyncIterator[T]

aggregate async 

aggregate(fn: Any) -> Any

take 

take(n: int) -> SQLValueStream[T]

drop 

drop(n: int) -> SQLValueStream[T]

distinct 

distinct() -> SQLValueStream[T]

sort 

sort[R: RichComparisonable](
    key: GetterCallable[T, R] | None = None, reverse: bool = False
) -> AsyncValueStreamable[T]

count async 

count() -> int

first async 

first() -> T | None

sum async 

sum[Item]() -> T | tuple[Item, ...] | None

sum[R: Valuable]() -> T | R | None

sum() -> Valuable | None

min async 

min[K: RichComparisonable](
    key: GetterCallable[T, K] | None = None,
) -> T | None

max async 

max[K: RichComparisonable](
    key: GetterCallable[T, K] | None = None,
) -> T | None