class Query[T](Iterable[T]):
@overload
def where[L, R](
self: Query[tuple[L, R]], condition: Expr[L] | Expr[R] | Expr[L | R], /
) -> Query[tuple[L, R]]: ...
@overload
def where(self, condition: Expr[T], /) -> Query[T]: ...
def where(self, condition: Expr[Any], /) -> Query[Any]:
return Where(self, condition)
@overload
def select[L, R, A](self: Query[tuple[L, R]], a: Field[L, A] | Field[R, A], /) -> Query[A]: ...
@overload
def select[L, R, A, B](
self: Query[tuple[L, R]], a: Field[L, A] | Field[R, A], b: Field[L, B] | Field[R, B], /
) -> Query[tuple[A, B]]: ...
@overload
def select[A](self, a: Field[T, A], /) -> Query[A]: ...
@overload
def select[A, B](self, a: Field[T, A], b: Field[T, B], /) -> Query[tuple[A, B]]: ...
@overload
def select[A, B, C](
self, a: Field[T, A], b: Field[T, B], c: Field[T, C], /
) -> Query[tuple[A, B, C]]: ...
def select(self, *fields: Field[Any, Any]) -> Query[Any]:
if not fields:
raise TypeError("select() needs at least one field")
return Select(self, fields)
def join[U](self, other: Query[U], on: JoinOn[T, U], /) -> Query[tuple[T, U]]:
return HashJoin(self, other, on)
@overload
def order_by[L, R](
self: Query[tuple[L, R]], *keys: SortKey[L] | SortKey[R]
) -> Query[tuple[L, R]]: ...
@overload
def order_by(self, *keys: SortKey[T]) -> Query[T]: ...
def order_by(self, *keys: SortKey[Any]) -> Query[Any]:
return OrderBy(self, keys)
@overload
def group_by[L, R, K](
self: Query[tuple[L, R]], key: Field[L, K] | Field[R, K], /
) -> Query[tuple[K]]: ...
@overload
def group_by[L, R, K, A](
self: Query[tuple[L, R]],
key: Field[L, K] | Field[R, K],
a: Aggregate[L, A] | Aggregate[R, A],
/,
) -> Query[tuple[K, A]]: ...
@overload
def group_by[L, R, K, A, B](
self: Query[tuple[L, R]],
key: Field[L, K] | Field[R, K],
a: Aggregate[L, A] | Aggregate[R, A],
b: Aggregate[L, B] | Aggregate[R, B],
/,
) -> Query[tuple[K, A, B]]: ...
@overload
def group_by[K](self, key: Field[T, K], /) -> Query[tuple[K]]: ...
@overload
def group_by[K, A](self, key: Field[T, K], a: Aggregate[T, A], /) -> Query[tuple[K, A]]: ...
@overload
def group_by[K, A, B](
self, key: Field[T, K], a: Aggregate[T, A], b: Aggregate[T, B], /
) -> Query[tuple[K, A, B]]: ...
def group_by(self, key: Field[Any, Any], *aggregates: Aggregate[Any, Any]) -> Query[Any]:
return GroupBy(self, key, aggregates)
def limit(self, n: int, offset: int = 0) -> Query[T]:
return Limit(self, n, offset)
def all(self) -> list[T]:
return list(self)
def first(self) -> T | None:
for row in self:
return row
return None
def count(self) -> int:
return sum(1 for _ in self)