Skip to content

Data API

DataR routes entity operations to persistence backends and shares their resources inside a context. Queries and streams describe and consume selections. For application-facing CRUD operations and schema projection, use DataS from fastsprout.actions. See DataS, DataR, and persistence for their relationship, or follow the SQLite tutorial to learn entity operations through a runnable example.

DataR

Bases: Routerable

dependencies class-attribute instance-attribute 

dependencies: Sequence[DependsClass] = ()

bindings instance-attribute 

bindings: Bindinger | None = None

bus instance-attribute 

bus: EventBus | None = bus

__init_bindings__ property 

__init_bindings__: Bindinger

__init__ 

__init__(
    *,
    bus: EventBus | None = None,
    dependencies: Sequence[DependsClass] | None = None,
) -> None

__aenter__ async 

__aenter__() -> Self

__aexit__ async 

__aexit__(
    exc_type: type[BaseException] | None,
    exc_value: BaseException | None,
    traceback: TracebackType | None,
) -> None

ability 

ability[E: Entitieable](
    entity: type[E],
) -> DataAbilitable[E, Any, BaseQuery[E, Any], Finalizable]

stream 

stream[E: Entitieable, ContextT, *Ts](
    q: JoinedQueryable[E, ContextT, *Ts],
) -> AsyncJoinedStreamable[*Ts,]

stream[E: Entitieable, StatementT](
    q: BaseQuery[E, StatementT],
) -> AsyncEntityStreamable[E]

stream[E: Entitieable, StatementT, ContextT, *Ts](
    q: BaseQuery[E, StatementT] | JoinedQueryable[E, ContextT, *Ts],
) -> AsyncEntityStreamable[E] | AsyncJoinedStreamable[*Ts,]

join 

join[E: Entitieable, StatementT](q: BaseQuery[E, StatementT]) -> RoutedJoin[E]

Start a lazy inner join; consume it inside this DataR context.

finalize async 

finalize() -> None

Flush staged intents, then finalize bound backends — highest priority first.

abort async 

abort() -> None

Drop staged intents and abort bound backends.

JoinedStream

Bases: AsyncJoinedStreamable[*Ts,]

Lazy inner joins producing flat, statically typed tuples.

Each step buffers its right source in a hash index, then streams the left rows in order. Duplicate keys produce every matching pair in right-source order. Keys must be hashable; None is an ordinary key. Plans are immutable, but replay requires replayable input sources. Routed sources must share one active DataR context; this is checked during iteration, including sources wrapped in stream transformations.

iterate instance-attribute 

iterate: Callable[[], AsyncGenerator[tuple[*Ts,], None]] = rows

__init__ 

__init__[T](first: AsyncIterable[T]) -> None

from_factory staticmethod 

from_factory[*Rs](
    factory: Callable[[], AsyncGenerator[tuple[*Rs,], None]],
) -> JoinedStream[*Rs,]

join 

join[T](
    stream: AsyncIterable[T], *, on: tuple[JoinKey[tuple[*Ts,]], JoinKey[T]]
) -> JoinedStream[*Ts, T]

Append a source, matching keys from the accumulated row and item.

Both key functions can be synchronous or asynchronous, following the same callback conventions as the other data streams.

__aiter__ 

__aiter__() -> AsyncGenerator[tuple[*Ts,], None]

Return an iterator; close it explicitly when stopping early.

to_list async 

to_list() -> list[tuple[*Ts,]]

SimpleAsyncEntityStream

Bases: BaseCommonStream[E], AsyncEntityStreamable[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]

concat async 

concat(*others: AsyncEntityStreamable[E] | AnyIterable[E])

map async 

map[R: Entitieable](mapper: MapperCallable[E, R])

sort async 

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

filter async 

filter(predicate: PredicateCallable[E])

only_of async 

only_of[R: Entitieable](entity: type[R])

distinct async 

distinct()

peek async 

peek(consumer: ConsumerCallable[E])

take async 

take(n: int)

drop async 

drop(n: int)

take_while async 

take_while(predicate: PredicateCallable[E])

drop_while async 

drop_while(predicate: PredicateCallable[E])

to_values async 

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

SimpleAsyncValueStream

Bases: BaseCommonStream[T], AsyncValueStreamable[T]

concat async 

concat(*others: AsyncValueStreamable[T] | AnyIterable[T])

map async 

map[R: Valuable](mapper: MapperCallable[T, R])

filter async 

filter(predicate: PredicateCallable[T])

distinct async 

distinct()

peek async 

peek(consumer: ConsumerCallable[T])

take async 

take(n: int)

drop async 

drop(n: int)

take_while async 

take_while(predicate: PredicateCallable[T])

drop_while async 

drop_while(predicate: PredicateCallable[T])

sort async 

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

sum async 

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

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

sum() -> Valuable | None

Relationship 

Relationship() -> Any

Declare an entity relation; SQL entities derive its foreign key.