How to Implement Upsert Operations in FastCRUD: A Complete Guide

FastCRUD provides upsert for single-record operations and upsert_multi for bulk operations, with native support for PostgreSQL, SQLite, and MySQL dialects through database-specific implementations in fastcrud/crud/database_specific.py.

To implement upsert operations in FastCRUD, you work with the FastCRUD class defined in fastcrud/crud/fast_crud.py. This library offers both single-record and bulk upsert capabilities that automatically handle the logic of inserting new records or updating existing ones based on primary key conflicts.

Understanding Upsert in FastCRUD

An upsert operation inserts a row when it does not exist or updates it when it does. FastCRUD implements this pattern through two primary methods:

• upsert – Handles a single record by attempting a get operation first, then calling either create or update based on existence.
• upsert_multi – Performs bulk upserts using native database syntax (ON CONFLICT for PostgreSQL/SQLite, ON DUPLICATE KEY UPDATE for MySQL) via dialect-specific builders in fastcrud/crud/database_specific.py.

Single-Record Upsert with FastCRUD.upsert

Method Signature and Workflow

The upsert method in fastcrud/crud/fast_crud.py (around line 1010) follows this signature:

async def upsert(
    self,
    db: AsyncSession,
    instance: UpdateSchemaType | CreateSchemaType,
    schema_to_select: type[SelectSchemaType] | None = None,
    return_as_model: bool = False,
) -> SelectSchemaType | dict[str, Any] | None:

Internal workflow:

1. Extract primary key values using _get_pk_dict.
2. Attempt to fetch the existing row via self.get.
3. If not found, create the record using self.create.
4. If found, update the record using self.update, then re-fetch to return the latest state.

Basic Usage Example

from fastcrud.crud.fast_crud import FastCRUD
from fastcrud.examples.user.model import User  # Your SQLAlchemy model

from fastcrud.examples.user.schemas import UserCreate, UserRead

crud = FastCRUD(User)

# session is an async SQLAlchemy AsyncSession

new_user = UserCreate(id=1, name="Alice", email="alice@example.com")
result = await crud.upsert(
    session, 
    new_user, 
    schema_to_select=UserRead, 
    return_as_model=True
)

# Returns UserRead(id=1, name="Alice", email="alice@example.com")

Handling Update Overrides

For PostgreSQL and SQLite, you can force specific column values during the update phase using update_override:

updated = await crud.upsert(
    session,
    UserCreate(id=1, name="Bob"),  # Input suggests "Bob"

    update_override={"name": "Robert"},  # Forces "Robert" on conflict

    schema_to_select=UserRead,
    return_as_model=True,
)

Important: MySQL does not support update_override or filters. Passing these parameters to a MySQL backend raises a ValueError.

Bulk Upsert Operations with FastCRUD.upsert_multi

Method Signature and Database Dialect Support

The upsert_multi method (starting around line 996 in fastcrud/crud/fast_crud.py) handles batch operations:

async def upsert_multi(
    self,
    db: AsyncSession,
    instances: list[UpdateSchemaType | CreateSchemaType],
    commit: bool = False,
    return_columns: list[str] | None = None,
    schema_to_select: type[SelectSchemaType] | None = None,
    return_as_model: bool = False,
    update_override: dict[str, Any] | None = None,
    **kwargs: Any,
) -> UpsertMultiResponseDict | UpsertMultiResponseModel[SelectSchemaType] | None:

Dialect dispatch logic inspects db.bind.dialect.name and delegates to specific builders in fastcrud/crud/database_specific.py:

Dialect	Helper Function	Native Syntax
PostgreSQL	upsert_multi_postgresql	ON CONFLICT (...) DO UPDATE ... RETURNING
SQLite	upsert_multi_sqlite	ON CONFLICT (...) DO UPDATE ... RETURNING
MySQL/MariaDB	upsert_multi_mysql	INSERT ... ON DUPLICATE KEY UPDATE

PostgreSQL and SQLite Bulk Upsert

These dialects support returning specific columns and applying filters to the conflict condition:

records = [
    UserCreate(id=1, name="Alice"),
    UserCreate(id=2, name="Bob"),
]

response = await crud.upsert_multi(
    session,
    records,
    commit=True,
    return_columns=["id", "name"],
    schema_to_select=UserRead,
    return_as_model=True,
)

# Response structure: {"created": [UserRead(...), ...], "updated": [...]}

print(response["created"])

The return_columns parameter triggers a RETURNING clause in the generated SQL. The response is formatted via format_multi_response in fastcrud/crud/fast_crud.py, producing either UpsertMultiResponseDict or UpsertMultiResponseModel instances defined in fastcrud/types.py.

MySQL Limitations and Workarounds

MySQL's INSERT ... ON DUPLICATE KEY UPDATE syntax lacks support for RETURNING clauses and conflict filters. FastCRUD enforces these constraints:

# This works: basic bulk upsert

await crud.upsert_multi(
    session,
    records,
    commit=True,
    update_override={"deleted_at": None},  # Resets soft-delete on conflict

)

Attempting to use unsupported features raises explicit errors:

# Raises: ValueError: MySQL does not support the returning clause...

await crud.upsert_multi(session, records, return_columns=["id"])

# Raises: ValueError if filters passed via **kwargs

await crud.upsert_multi(session, records, name__like="A%")

Return Types and Response Handling

FastCRUD structures bulk upsert responses using types defined in fastcrud/types.py:

• UpsertMultiResponseDict – A typed dictionary with "created" and "updated" keys containing lists of dictionaries.
• UpsertMultiResponseModel[SelectSchemaType] – A generic model where the lists contain Pydantic model instances when return_as_model=True.

The internal format_multi_response method processes raw database results into these structures, separating created rows from updated rows based on the operation metadata.

Summary

• Single-record upserts use FastCRUD.upsert in fastcrud/crud/fast_crud.py, which performs a get-then-create-or-update workflow.
• Bulk upserts use FastCRUD.upsert_multi, which delegates to dialect-specific builders in fastcrud/crud/database_specific.py for PostgreSQL, SQLite, and MySQL.
• PostgreSQL and SQLite support return_columns, update_override, and conflict filters via ON CONFLICT syntax.
• MySQL supports basic upserts via ON DUPLICATE KEY UPDATE but raises ValueError for returning clauses or filters.
• Response types UpsertMultiResponseDict and UpsertMultiResponseModel structure bulk operation results, defined in fastcrud/types.py.

Frequently Asked Questions

What is the difference between upsert and upsert_multi in FastCRUD?

upsert handles a single record at a time by first attempting to fetch the existing row via get, then calling either create or update depending on existence. upsert_multi performs batch operations using native database upsert syntax (ON CONFLICT or ON DUPLICATE KEY UPDATE) through dialect-specific implementations in fastcrud/crud/database_specific.py, making it significantly more efficient for large datasets.

Why does MySQL raise a ValueError when using return_columns with upsert_multi?

MySQL's INSERT ... ON DUPLICATE KEY UPDATE syntax does not support a RETURNING clause, unlike PostgreSQL and SQLite. When you pass return_columns to upsert_multi with a MySQL backend, FastCRUD explicitly raises ValueError: MySQL does not support the returning clause for insert operations to prevent runtime SQL errors. For MySQL, omit return_columns and perform a separate query if you need the inserted/updated data.

How do I force specific column values during a conflict update in FastCRUD?

Use the update_override parameter available in both upsert and upsert_multi methods. Pass a dictionary mapping column names to desired values, such as update_override={"updated_at": datetime.now(), "deleted_at": None}. FastCRUD merges these overrides with the standard column mappings when generating the ON CONFLICT DO UPDATE or ON DUPLICATE KEY UPDATE SQL. Note that this feature is only supported for PostgreSQL and SQLite; MySQL will raise a ValueError if update_override is provided.

Can I apply filters to the upsert conflict condition in FastCRUD?

Yes, but only when using PostgreSQL or SQLite. You can pass filter arguments (such as name__like="A%") as **kwargs to upsert_multi. FastCRUD processes these through _filter_processor.parse_filters and applies them as a WHERE clause to the ON CONFLICT DO UPDATE statement. This allows conditional updates based on existing data states. MySQL does not support this functionality; passing filters with a MySQL backend results in a ValueError.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}