Skip to content

upsert

Upsert operations for database controllers.

Classes:

Classes

UpsertController

Python
UpsertController(model: type[ModelT], db: DatabaseService)

Handles upsert and get-or-create operations.

Initialize the upsert controller.

Parameters:

  • model (type[ModelT]) –

    The SQLModel to perform upsert operations on.

  • db (DatabaseService) –

    The database service instance.

Returned by:

Methods:

Functions

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.

Raises:

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.