# How to Implement Upsert Operations in FastCRUD: A Complete Guide

> Learn how to implement upsert operations in FastCRUD for single and multi-record updates. This guide covers native support for PostgreSQL, SQLite, and MySQL.

- Repository: [Benav Labs/fastcrud](https://github.com/benavlabs/fastcrud)
- Tags: how-to-guide
- Published: 2026-02-26

---

**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`](https://github.com/benavlabs/fastcrud/blob/main/fastcrud/crud/database_specific.py).**

To implement upsert operations in FastCRUD, you work with the `FastCRUD` class defined in [`fastcrud/crud/fast_crud.py`](https://github.com/benavlabs/fastcrud/blob/main/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`](https://github.com/benavlabs/fastcrud/blob/main/fastcrud/crud/database_specific.py).

## Single-Record Upsert with `FastCRUD.upsert`

### Method Signature and Workflow

The `upsert` method in [`fastcrud/crud/fast_crud.py`](https://github.com/benavlabs/fastcrud/blob/main/fastcrud/crud/fast_crud.py) (around line 1010) follows this signature:

```python
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

```python
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`:

```python
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`](https://github.com/benavlabs/fastcrud/blob/main/fastcrud/crud/fast_crud.py)) handles batch operations:

```python
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`](https://github.com/benavlabs/fastcrud/blob/main/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:

```python
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`](https://github.com/benavlabs/fastcrud/blob/main/fastcrud/crud/fast_crud.py), producing either `UpsertMultiResponseDict` or `UpsertMultiResponseModel` instances defined in [`fastcrud/types.py`](https://github.com/benavlabs/fastcrud/blob/main/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:

```python

# 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:

```python

# 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`](https://github.com/benavlabs/fastcrud/blob/main/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`](https://github.com/benavlabs/fastcrud/blob/main/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`](https://github.com/benavlabs/fastcrud/blob/main/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`](https://github.com/benavlabs/fastcrud/blob/main/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`](https://github.com/benavlabs/fastcrud/blob/main/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`.