Entity Framework (orm)

pydantic_linq.orm.DbContext

DbContext(bind: Engine | AsyncEngine | Session | AsyncSession)

with DbContext(engine) as db: db.set(Hero).where(lambda h: h.age > 30).to_list().

Pass an AsyncEngine (or AsyncSession) and use async with and the *_async methods for async I/O. As in EF, queries don't flush pending changes (no autoflush) and entities keep their values after save_changes.

session instance-attribute

session = bind

add

add(entity: object) -> None

add_range

add_range(entities: Iterable[object]) -> None

attach

attach(entity: object) -> None

Attach: track an entity loaded elsewhere, as unchanged.

begin_transaction

begin_transaction() -> SessionTransaction

Database.BeginTransaction: call commit() or rollback() on the result, or use it in a with block.

begin_transaction_async async

begin_transaction_async() -> Any

clear

clear() -> None

ChangeTracker.Clear: stop tracking every entity.

ensure_created

ensure_created() -> None

Database.EnsureCreated: create the tables of every SQLModel table model that don't exist yet.

ensure_created_async async

ensure_created_async() -> None

ensure_deleted

ensure_deleted() -> None

Database.EnsureDeleted: drop the tables of every SQLModel table model.

ensure_deleted_async async

ensure_deleted_async() -> None

execute_sql

execute_sql(sql: str, **params: Any) -> int

Database.ExecuteSql: run a statement; returns the affected row count.

execute_sql_async async

execute_sql_async(sql: str, **params: Any) -> int

find

find(model: type[T], *key: Any) -> T | None

Find: by primary key; returns the tracked entity without a query when it is already loaded.

find_async async

find_async(model: type[T], *key: Any) -> T | None

from_sql

from_sql(model: type[T], sql: str, **params: Any) -> Queryable[T]

FromSql: a query over raw SQL that returns the model's columns. Bind values as :name params, never by formatting them into the string. LINQ operators after it compose over it as a subquery.

remove

remove(entity: object) -> None

remove_range

remove_range(entities: Iterable[object]) -> None

save_changes

save_changes() -> int

SaveChanges: commit; returns the number of entities written.

save_changes_async async

save_changes_async() -> int

set

set(model: type[T]) -> Queryable[T]

Set() / DbSet: the root query over a table.

update

update(entity: object) -> None

Update: save the detached entity's current values on the next save_changes.

update_range

update_range(entities: Iterable[object]) -> None

pydantic_linq.orm.Queryable

Queryable(session: AnySession, model: type[T])

Bases: Enumerable[T]

A query against the database. Nothing runs until iteration or a terminal operator.

all

all(predicate: Callable[[T], Any]) -> bool

all_async async

all_async(predicate: Callable[[T], Any]) -> bool

any

any(predicate: Callable[[T], Any] | None = None) -> bool

any_async async

any_async(predicate: Callable[[T], Any] | None = None) -> bool

as_async_enumerable

as_async_enumerable() -> AsyncEnumerable[T]

AsAsyncEnumerable: continue with the System.Linq.Async operators over the results.

as_enumerable

as_enumerable() -> Enumerable[T]

AsEnumerable: later operators run in memory. The query runs on each enumeration.

as_no_tracking

as_no_tracking() -> Queryable[T]

AsNoTracking: returned entities are detached, so changes to them aren't saved.

as_single_query

as_single_query() -> Queryable[T]

AsSingleQuery: includes load through LEFT OUTER JOINs in the main query.

as_split_query

as_split_query() -> Queryable[T]

AsSplitQuery: each include loads in its own SELECT ... IN query (the default here).

as_tracking

as_tracking() -> Queryable[T]

average

average(selector: Callable[[T], Any] | None = None) -> float

average_async async

average_async(selector: Callable[[T], Any] | None = None) -> float

concat

concat(second: Iterable[T]) -> Queryable[T]

contains

contains(value: T) -> bool

contains_async async

contains_async(value: T) -> bool

count

count(predicate: Callable[[T], Any] | None = None) -> int

count_async async

count_async(predicate: Callable[[T], Any] | None = None) -> int

count_by

count_by(key_selector: Callable[[T], K]) -> Queryable[tuple[K, int]]

distinct

distinct() -> Queryable[T]

element_at

element_at(index: int) -> T

element_at_async async

element_at_async(index: int) -> T

element_at_or_default

element_at_or_default(index: int, default: T | None = None) -> T | None

element_at_or_default_async async

element_at_or_default_async(index: int, default: T | None = None) -> T | None

except_

except_(second: Iterable[T]) -> Queryable[T]

execute_delete

execute_delete() -> int

ExecuteDelete: one DELETE for the matching rows; returns the row count. Tracked entities aren't updated. The statement runs in the session's transaction, so save_changes() commits it.

execute_delete_async async

execute_delete_async() -> int

execute_update

execute_update(setters: Callable[[T], dict[Any, Any]]) -> int

ExecuteUpdate: execute_update(lambda h: {h.age: h.age + 1}) is one UPDATE; returns the row count.

execute_update_async async

execute_update_async(setters: Callable[[T], dict[Any, Any]]) -> int

first

first(predicate: Callable[[T], Any] | None = None) -> T

first_async async

first_async(predicate: Callable[[T], Any] | None = None) -> T

first_or_default

first_or_default(predicate: Callable[[T], Any] | None = None, default: T | None = None) -> T | None

first_or_default_async async

first_or_default_async(predicate: Callable[[T], Any] | None = None, default: T | None = None) -> T | None

for_each_async async

for_each_async(action: Callable[[T], Any]) -> None

ForEachAsync: action may be sync or async.

group_by

group_by(key_selector: Callable[[T], K]) -> GroupedQueryable[K, T]

group_by(key_selector: Callable[[T], K], element_selector: Callable[[T], E]) -> GroupedQueryable[K, E]

group_by(key_selector: Callable[[T], K], element_selector: Callable[[T], E] | None = None, *, result_selector: Callable[[K, Enumerable[E]], R]) -> Queryable[R]

group_by(key_selector: Callable[[T], Any], element_selector: Callable[[T], Any] | None = None, *, result_selector: Callable[[Any, Enumerable[Any]], Any] | None = None) -> Any

GroupBy. Followed by select / where / order_by over g.key, g.count(), g.sum(...), ... it runs as GROUP BY / HAVING. Enumerated directly, the groups are built in memory.

include

include(navigation: Callable[[T], Any]) -> Queryable[T]

Include: eager-load a relationship (lambda h: h.team).

intersect

intersect(second: Iterable[T]) -> Queryable[T]

join

join(inner: Iterable[I], outer_key_selector: Callable[[T], K], inner_key_selector: Callable[[I], K], result_selector: Callable[[T, I], R]) -> Queryable[R]

Join with another Queryable is an INNER JOIN.

last

last(predicate: Callable[[T], Any] | None = None) -> T

last_async async

last_async(predicate: Callable[[T], Any] | None = None) -> T

last_or_default

last_or_default(predicate: Callable[[T], Any] | None = None, default: T | None = None) -> T | None

last_or_default_async async

last_or_default_async(predicate: Callable[[T], Any] | None = None, default: T | None = None) -> T | None

left_join

left_join(inner: Iterable[I], outer_key_selector: Callable[[T], K], inner_key_selector: Callable[[I], K], result_selector: Callable[[T, I | None], R]) -> Queryable[R]

LeftJoin (.NET 10 / EF 10): LEFT OUTER JOIN; unmatched rows pair with None.

load

load() -> None

Load: run the query so its entities are tracked, without returning them.

load_async async

load_async() -> None

long_count

long_count(predicate: Callable[[T], Any] | None = None) -> int

long_count_async async

long_count_async(predicate: Callable[[T], Any] | None = None) -> int

max

max(selector: Callable[[T], Any] | None = None) -> Any

max_async async

max_async(selector: Callable[[T], Any] | None = None) -> Any

max_by

max_by(key_selector: Callable[[T], Any]) -> T

max_by_async async

max_by_async(key_selector: Callable[[T], Any]) -> T

min

min(selector: Callable[[T], Any] | None = None) -> Any

min_async async

min_async(selector: Callable[[T], Any] | None = None) -> Any

min_by

min_by(key_selector: Callable[[T], Any]) -> T

min_by_async async

min_by_async(key_selector: Callable[[T], Any]) -> T

order

order(comparer: Comparer | None = None) -> OrderedQueryable[T]

order_by

order_by(key_selector: Callable[[T], Any], comparer: Comparer | None = None) -> OrderedQueryable[T]

order_by_descending

order_by_descending(key_selector: Callable[[T], Any], comparer: Comparer | None = None) -> OrderedQueryable[T]

order_descending

order_descending(comparer: Comparer | None = None) -> OrderedQueryable[T]

reverse

reverse() -> Queryable[T]

right_join

right_join(inner: Iterable[I], outer_key_selector: Callable[[T], K], inner_key_selector: Callable[[I], K], result_selector: Callable[[T | None, I], R]) -> Queryable[R]

RightJoin (.NET 10 / EF 10): run as the inner query LEFT JOIN this one.

select

select(selector: Callable[[T], U]) -> Queryable[U]

Columns (lambda h: h.name, lambda h: (h.id, h.name)) are selected in SQL; anything else runs in memory.

select_many

select_many(selector: Callable[[T], Iterable[U]]) -> Queryable[U]

select_many(selector: Callable[[T], Iterable[C]], result_selector: Callable[[T, C], U]) -> Queryable[U]

select_many(selector: Callable[[T], Iterable[Any]], result_selector: Callable[[T, Any], Any] | None = None) -> Queryable[Any]

SelectMany over a collection navigation (lambda t: t.heroes) is a JOIN; anything else runs in memory.

single

single(predicate: Callable[[T], Any] | None = None) -> T

single_async async

single_async(predicate: Callable[[T], Any] | None = None) -> T

single_or_default

single_or_default(predicate: Callable[[T], Any] | None = None, default: T | None = None) -> T | None

single_or_default_async async

single_or_default_async(predicate: Callable[[T], Any] | None = None, default: T | None = None) -> T | None

skip

skip(count: int) -> Queryable[T]

sum

sum(selector: Callable[[T], Any] | None = None) -> Any

sum_async async

sum_async(selector: Callable[[T], Any] | None = None) -> Any

tag_with

tag_with(tag: str) -> Queryable[T]

TagWith: put a comment into the SQL, to find the query in logs.

take

take(count: int | slice) -> Queryable[T]

Take(int) or Take(Range); a range with negative bounds or a step runs in memory.

then_include

then_include(navigation: Callable[[Any], Any]) -> Queryable[T]

ThenInclude: eager-load a relationship of the entity the previous include loaded.

to_array_async async

to_array_async() -> list[T]

to_async_enumerable

to_async_enumerable() -> AsyncEnumerable[T]

to_dictionary_async async

to_dictionary_async(key_selector: Callable[[T], K], element_selector: Callable[[T], Any] | None = None) -> dict[K, Any]

to_hash_set_async async

to_hash_set_async() -> set[T]

to_list

to_list() -> list[T]

to_list_async async

to_list_async() -> list[T]

to_lookup_async async

to_lookup_async(key_selector: Callable[[T], K]) -> Lookup[K, T]

to_query_string

to_query_string() -> str

ToQueryString: the SQL this query runs.

try_get_non_enumerated_count

try_get_non_enumerated_count() -> int | None

union

union(second: Iterable[T]) -> Queryable[T]

where

where(predicate: Callable[[T], Any]) -> Queryable[T]

pydantic_linq.orm.OrderedQueryable

OrderedQueryable(session: AnySession, model: type[T])

Bases: Queryable[T], OrderedEnumerable[T]

IOrderedQueryable: then_by / then_by_descending add ORDER BY columns.

then_by

then_by(key_selector: Callable[[T], Any], comparer: Comparer | None = None) -> OrderedQueryable[T]

then_by_descending

then_by_descending(key_selector: Callable[[T], Any], comparer: Comparer | None = None) -> OrderedQueryable[T]

pydantic_linq.orm.GroupedQueryable

GroupedQueryable(query: Queryable[Any], key_selector: Callable[[Any], K], element_selector: Callable[[Any], Any] | None, having: tuple[Callable[[Any], Any], ...] = (), order: tuple[tuple[Callable[[Any], Any], bool, Comparer | None], ...] = ())

Bases: Enumerable[Grouping[K, T]]

Lambdas get g with g.key and the aggregates g.count(), g.sum(sel), g.min(sel), g.max(sel), g.average(sel), g.any(). The same lambdas work on in-memory Grouping objects, so a select that can't be translated (or plain enumeration) groups in memory instead.

count

count(predicate: Callable[[Grouping[K, T]], Any] | None = None) -> int

The number of groups.

count_async async

count_async(predicate: Callable[[Grouping[K, T]], Any] | None = None) -> int

long_count

long_count(predicate: Callable[[Grouping[K, T]], Any] | None = None) -> int

order_by

order_by(key_selector: Callable[[Grouping[K, T]], Any], comparer: Comparer | None = None) -> OrderedGroupedQueryable[K, T]

order_by_descending

order_by_descending(key_selector: Callable[[Grouping[K, T]], Any], comparer: Comparer | None = None) -> OrderedGroupedQueryable[K, T]

select

select(selector: Callable[[Grouping[K, T]], R]) -> Queryable[R]

to_list_async async

to_list_async() -> list[Grouping[K, T]]

where

where(predicate: Callable[[Grouping[K, T]], Any]) -> GroupedQueryable[K, T]

HAVING.

pydantic_linq.orm.SqlGrouping

SqlGrouping(key: Any, element: Any)

What a lambda after group_by receives: g.key and SQL aggregates over the group's elements.

key instance-attribute

key = key

any

any(predicate: Callable[[Any], Any] | None = None) -> Any

average

average(selector: Callable[[Any], Any] | None = None) -> Any

count

count(predicate: Callable[[Any], Any] | None = None) -> Any

long_count

long_count(predicate: Callable[[Any], Any] | None = None) -> Any

max

max(selector: Callable[[Any], Any] | None = None) -> Any

min

min(selector: Callable[[Any], Any] | None = None) -> Any

sum

sum(selector: Callable[[Any], Any] | None = None) -> Any