base ¶

Database controller components for modular database operations.

Modules:

base_controller –

Main BaseController that composes all specialized controllers with lazy initialization.
bulk –

Bulk operations for database controllers.
crud –

Core CRUD operations for database controllers.
filters –

Shared filter utilities for database controllers.
pagination –

Pagination operations for database controllers.
performance –

Performance analysis for database controllers.
query –

Query operations for database controllers.
transaction –

Transaction management for database controllers.
upsert –

Upsert operations for database controllers.

Classes:

BaseController –

Composed database controller with lazy-loaded specialized operations.

Classes¶

BaseController ¶

Python

BaseController(model: type[ModelT], db: DatabaseService | None = None)

Composed database controller with lazy-loaded specialized operations.

This controller delegates operations to specialized controllers while maintaining backward compatibility with the original BaseController API. Core CRUD and Query controllers are eagerly initialized, while specialized controllers (pagination, bulk, transaction, performance, upsert) use lazy initialization to reduce overhead for simple use cases.

Initialize the base controller with lazy-loaded specialized controllers.

Parameters:

model (type[ModelT]) –

The SQLModel class to perform operations on.
db (DatabaseService | None, default: None ) –

The database service instance. Must be provided.

Raises:

RuntimeError –

If db is None, as database service is required.

Returned by:

utils Functions create_enhanced_controller_from

Subclassed by:

Methods:

create –

Create a new record.
get_by_id –

Get a record by ID.
update_by_id –

Update a record by ID.
delete_by_id –

Delete a record by ID.
exists –

Check if a record exists.
find_one –

Find one record.
find_all –

Find all records with performance optimizations.
find_all_with_options –

Find all records with relationship loading options.
count –

Count records.
get_all –

Get all records (alias for find_all without pagination).
execute_query –

Execute a custom query.
find_with_json_query –

Find records using JSON column queries.
find_with_array_contains –

Find records where array column contains value.
find_with_full_text_search –

Find records using full-text search.
paginate –

Paginate records with metadata.
find_paginated –

Find paginated records with relationship loading.
bulk_create –

Create multiple records in bulk.
bulk_update –

Update multiple records in bulk.
bulk_delete –

Delete multiple records in bulk.
update_where –

Update records matching filters.
delete_where –

Delete records matching filters.
bulk_upsert_with_conflict_resolution –

Bulk upsert with conflict resolution.
with_session –

Execute operation within a session context.
with_transaction –

Execute operation within a transaction context.
execute_transaction –

Execute a callback within a transaction.
get_table_statistics –

Get comprehensive table statistics.
explain_query_performance –

Explain query performance with optional analysis.
upsert_by_field –

Upsert a record by a specific field.
upsert_by_id –

Upsert a record by ID.
get_or_create_by_field –

Get existing record or create new one by field.
get_or_create –

Get existing record or create new one.
upsert –

Upsert a record.

Attributes:

db_service (DatabaseService) –

Database service property for test compatibility.
model_class (type[ModelT]) –

Model class property for test compatibility.

Attributes¶

db_service `property` ¶

Python

db_service: DatabaseService

Database service property for test compatibility.

Returns:

DatabaseService –

The database service instance.

model_class `property` ¶

Python

model_class: type[ModelT]

Model class property for test compatibility.

Returns:

type[ModelT] –

The SQLModel class.

Functions¶

_get_pagination ¶

Python

_get_pagination() -> PaginationController[ModelT]

Get or create pagination controller.

Returns:

PaginationController[ModelT] –

The pagination controller instance.

_get_bulk ¶

Python

_get_bulk() -> BulkOperationsController[ModelT]

Get or create bulk operations controller.

Returns:

BulkOperationsController[ModelT] –

The bulk operations controller instance.

_get_transaction ¶

Python

_get_transaction() -> TransactionController[ModelT]

Get or create transaction controller.

Returns:

TransactionController[ModelT] –

The transaction controller instance.

_get_performance ¶

Python

_get_performance() -> PerformanceController[ModelT]

Get or create performance controller.

Returns:

PerformanceController[ModelT] –

The performance controller instance.

_get_upsert ¶

Python

_get_upsert() -> UpsertController[ModelT]

Get or create upsert controller.

Returns:

UpsertController[ModelT] –

The upsert controller instance.

create `async` ¶

Python

create(**kwargs: Any) -> ModelT

Create a new record.

Returns:

ModelT –

The newly created record.

get_by_id `async` ¶

Python

get_by_id(record_id: Any) -> ModelT | None

Get a record by ID.

Returns:

ModelT | None –

The record if found, None otherwise.

update_by_id `async` ¶

Python

update_by_id(record_id: Any, **values: Any) -> ModelT | None

Update a record by ID.

Returns:

ModelT | None –

The updated record, or None if not found.

delete_by_id `async` ¶

Python

delete_by_id(record_id: Any) -> bool

Delete a record by ID.

Returns:

bool –

True if deleted successfully, False otherwise.

exists `async` ¶

Python

exists(filters: Any) -> bool

Check if a record exists.

Returns:

bool –

True if record exists, False otherwise.

find_one `async` ¶

Python

find_one(filters: Any | None = None, order_by: Any | None = None) -> ModelT | None

Find one record.

Returns:

ModelT | None –

The found record, or None if not found.

find_all `async` ¶

Python

find_all(
    filters: Any | None = None,
    order_by: Any | None = None,
    limit: int | None = None,
    offset: int | None = None,
) -> list[ModelT]

Find all records with performance optimizations.

Returns:

list[ModelT] –

List of found records.

find_all_with_options `async` ¶

Python

find_all_with_options(
    filters: Any | None = None,
    order_by: Any | None = None,
    limit: int | None = None,
    offset: int | None = None,
    load_relationships: list[str] | None = None,
) -> list[ModelT]

Find all records with relationship loading options.

Returns:

list[ModelT] –

List of found records with loaded relationships.

count `async` ¶

Python

count(filters: Any | None = None) -> int

Count records.

Returns:

int –

The count of matching records.

get_all `async` ¶

Python

get_all(filters: Any | None = None, order_by: Any | None = None) -> list[ModelT]

Get all records (alias for find_all without pagination).

Returns:

list[ModelT] –

List of all matching records.

execute_query `async` ¶

Python

execute_query(query: Any) -> Any

Execute a custom query.

Returns:

Any –

The query result.

find_with_json_query `async` ¶

Python

find_with_json_query(
    json_column: str, json_path: str, value: Any, filters: Any | None = None
) -> list[ModelT]

Find records using JSON column queries.

Returns:

list[ModelT] –

List of records matching the JSON query.

find_with_array_contains `async` ¶

Python

find_with_array_contains(
    array_column: str, value: Any, filters: Any | None = None
) -> list[ModelT]

Find records where array column contains value.

Returns:

list[ModelT] –

List of records with matching array values.

find_with_full_text_search `async` ¶

Python

find_with_full_text_search(
    search_columns: list[str], search_term: str, filters: Any | None = None
) -> list[ModelT]

Find records using full-text search.

Returns:

list[ModelT] –

List of records matching the search term.

paginate `async` ¶

Python

paginate(
    page: int = 1,
    per_page: int = 20,
    filters: Any | None = None,
    order_by: Any | None = None,
) -> PaginationResult[ModelT]

Paginate records with metadata.

Returns:

PaginationResult[ModelT] –

Pagination result with items, total, and page info.

find_paginated `async` ¶

Python

find_paginated(
    page: int = 1,
    per_page: int = 20,
    filters: Any | None = None,
    order_by: Any | None = None,
    load_relationships: list[str] | None = None,
) -> PaginationResult[ModelT]

Find paginated records with relationship loading.

Returns:

PaginationResult[ModelT] –

Pagination result with items and relationships loaded.

bulk_create `async` ¶

Python

bulk_create(items: list[dict[str, Any]]) -> list[ModelT]

Create multiple records in bulk.

Returns:

list[ModelT] –

List of created records.

bulk_update `async` ¶

Python

bulk_update(updates: list[tuple[Any, dict[str, Any]]]) -> int

Update multiple records in bulk.

Returns:

int –

Number of records updated.

bulk_delete `async` ¶

Python

bulk_delete(record_ids: list[Any]) -> int

Delete multiple records in bulk.

Returns:

int –

Number of records deleted.

update_where `async` ¶

Python

update_where(filters: Any, values: dict[str, Any]) -> int

Update records matching filters.

Returns:

int –

Number of records updated.

delete_where `async` ¶

Python

delete_where(filters: Any) -> int

Delete records matching filters.

Returns:

int –

Number of records deleted.

bulk_upsert_with_conflict_resolution `async` ¶

Python

bulk_upsert_with_conflict_resolution(
    items: list[dict[str, Any]],
    conflict_columns: list[str],
    update_columns: list[str] | None = None,
) -> list[ModelT]

Bulk upsert with conflict resolution.

Returns:

list[ModelT] –

List of upserted records.

with_session `async` ¶

Python

with_session[R](operation: Callable[[Any], Awaitable[R]]) -> R

Execute operation within a session context.

Returns:

R –

The result of the operation.

with_transaction `async` ¶

Python

with_transaction[R](operation: Callable[[Any], Awaitable[R]]) -> R

Execute operation within a transaction context.

Returns:

R –

The result of the operation.

execute_transaction `async` ¶

Python

execute_transaction(callback: Callable[[], Any]) -> Any

Execute a callback within a transaction.

Returns:

Any –

The result of the callback.

get_table_statistics `async` ¶

Python

get_table_statistics() -> dict[str, Any]

Get comprehensive table statistics.

Returns:

dict[str, Any] –

Dictionary containing table statistics.

explain_query_performance `async` ¶

Python

explain_query_performance(
    query: Any, analyze: bool = False, buffers: bool = False
) -> dict[str, Any]

Explain query performance with optional analysis.

Returns:

dict[str, Any] –

Dictionary containing query execution plan and statistics.

upsert_by_field `async` ¶

Python

upsert_by_field(
    field_name: str,
    field_value: Any,
    defaults: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[ModelT, bool]

Upsert a record by a specific field.

Returns:

tuple[ModelT, bool] –

Tuple of (record, created) where created is True if new record was created.

upsert_by_id `async` ¶

Python

upsert_by_id(
    record_id: Any, defaults: dict[str, Any] | None = None, **kwargs: Any
) -> tuple[ModelT, bool]

Upsert a record by ID.

Returns:

tuple[ModelT, bool] –

Tuple of (record, created) where created is True if new record was created.

get_or_create_by_field `async` ¶

Python

get_or_create_by_field(
    field_name: str,
    field_value: Any,
    defaults: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[ModelT, bool]

Get existing record or create new one by field.

Returns:

tuple[ModelT, bool] –

Tuple of (record, created) where created is True if new record was created.

get_or_create `async` ¶

Python

get_or_create(
    defaults: dict[str, Any] | None = None, **filters: Any
) -> tuple[ModelT, bool]

Get existing record or create new one.

Returns:

tuple[ModelT, bool] –

Tuple of (record, created) where created is True if new record was created.

upsert `async` ¶

Python

upsert(
    filters: dict[str, Any], defaults: dict[str, Any] | None = None, **kwargs: Any
) -> tuple[ModelT, bool]

Upsert a record.

Returns:

tuple[ModelT, bool] –

Tuple of (record, created) where created is True if new record was created.

base ¶

Classes¶

BaseController ¶

Attributes¶

db_service property ¶

model_class property ¶

Functions¶

_get_pagination ¶

_get_bulk ¶

_get_transaction ¶

_get_performance ¶

_get_upsert ¶

create async ¶

get_by_id async ¶

update_by_id async ¶

delete_by_id async ¶

exists async ¶

find_one async ¶

find_all async ¶

find_all_with_options async ¶

count async ¶

get_all async ¶

execute_query async ¶

find_with_json_query async ¶

find_with_array_contains async ¶

find_with_full_text_search async ¶

paginate async ¶

find_paginated async ¶

bulk_create async ¶

bulk_update async ¶

bulk_delete async ¶

update_where async ¶

delete_where async ¶

bulk_upsert_with_conflict_resolution async ¶

with_session async ¶

with_transaction async ¶

execute_transaction async ¶

get_table_statistics async ¶

explain_query_performance async ¶

upsert_by_field async ¶

upsert_by_id async ¶

get_or_create_by_field async ¶

get_or_create async ¶

upsert async ¶

db_service `property` ¶

model_class `property` ¶

create `async` ¶

get_by_id `async` ¶

update_by_id `async` ¶

delete_by_id `async` ¶

exists `async` ¶

find_one `async` ¶

find_all `async` ¶

find_all_with_options `async` ¶

count `async` ¶

get_all `async` ¶

execute_query `async` ¶

find_with_json_query `async` ¶

find_with_array_contains `async` ¶

find_with_full_text_search `async` ¶

paginate `async` ¶

find_paginated `async` ¶

bulk_create `async` ¶

bulk_update `async` ¶

bulk_delete `async` ¶

update_where `async` ¶

delete_where `async` ¶

bulk_upsert_with_conflict_resolution `async` ¶

with_session `async` ¶

with_transaction `async` ¶

execute_transaction `async` ¶

get_table_statistics `async` ¶

explain_query_performance `async` ¶

upsert_by_field `async` ¶

upsert_by_id `async` ¶

get_or_create_by_field `async` ¶

get_or_create `async` ¶

upsert `async` ¶