Mutations¶

Creating, updating, and deleting records. All mutations are executed by the Rust engine, and all of them participate in an active transaction automatically.

Ferro's write surface is three verbs with three distinct intents — none of them ever silently does more work than you asked for:

Creating Records¶

create¶

Model.create(**fields) validates, inserts, and returns the persisted instance in one call. For inserting many rows, Model.bulk_create(instances) batches them into a single statement and returns the inserted count:

        alice = await Author.create(name="Alice", email="[email protected]")

        post = await Post.create(
            title="Why Ferro is Fast",
            body="Ferro hands SQL generation and row hydration to a Rust engine...",
            published=True,
            author=alice,
        )

        # Insert many rows in a single statement
        await Post.bulk_create(
            [
                Post(title="Async Patterns", body="...", published=True, author_id=alice.id),
                Post(title="Unfinished Draft", body="...", author_id=alice.id),
            ]
        )

create() is a plain INSERT — it never updates an existing row. A duplicate primary key or unique value raises UniqueViolationError, and the existing row is left untouched:

        alice = await Customer.get(customer.id)

        # create() never updates an existing row — a duplicate is an error
        try:
            await Customer.create(id=alice.id, email="[email protected]")
        except UniqueViolationError:
            ...  # the existing row is untouched

Reach for upsert() when you want insert-or-update semantics.

Pass related instances directly (author=alice) or set the shadow foreign-key column (author_id=alice.id) — see Relationships. Pydantic validation runs when each instance is constructed, so invalid data raises before the database is touched.

Defaults¶

Fields with default or default_factory fill themselves in:

from datetime import datetime

from ferro import Field, Model


class Article(Model):
    id: int | None = Field(default=None, primary_key=True)
    title: str
    draft: bool = True
    created_at: datetime = Field(default_factory=datetime.now)

from datetime import datetime
from typing import Annotated

from ferro import Field, Model


class Article(Model):
    id: Annotated[int | None, Field(default=None, primary_key=True)]
    title: str
    draft: bool = True
    created_at: Annotated[datetime, Field(default_factory=datetime.now)]

article = await Article.create(title="Hello")
# article.draft is True, article.created_at is set

Saving: INSERT or UPDATE¶

Every instance is either transient (constructed with Model(...) and never persisted) or persisted (fetched from the database, or successfully saved). save() uses that state to decide what to do:

• a transient instance is INSERTed — a duplicate primary key or unique value raises UniqueViolationError, exactly like create();
• a persisted instance is UPDATEd by primary key (UPDATE ... WHERE pk = ?).

        bob = Customer(email="[email protected]", name="Bob")
        await bob.save()  # never persisted: INSERT (bob.id is now set)

        bob.plan = "pro"
        await bob.save()  # persisted: UPDATE ... WHERE id = ?

        bobs = await Customer.where(lambda customer: customer.name == "Bob").count()
        assert bobs == 1  # two saves, one row

instance.delete() returns the instance to transient, so a subsequent save() INSERTs a new row. refresh() keeps it persisted.

If the row behind a persisted instance no longer exists — deleted by another writer, or you mutated the primary-key field before saving — the UPDATE matches nothing and save() raises ModelDoesNotExist rather than silently resurrecting the row.

Upsert¶

Model.upsert(**fields) inserts the row, or updates the existing row when the primary key conflicts:

        # No row with this primary key: INSERT
        carol = await Customer.upsert(id=100, email="[email protected]", name="Carol")

        # Same primary key again: UPDATE of the existing row
        carol = await Customer.upsert(id=100, email="[email protected]", plan="pro")
        assert carol.plan == "pro"
        assert await Customer.where(lambda customer: customer.id == 100).count() == 1

        # The whole row is written: `name` was left unset above, so the stored
        # "Carol" reverted to the field default — pass every field you care about.
        stored = await Customer.get(100)
        assert stored.name == ""

Things to know:

• The conflict target is the primary key only. A conflict on some other unique column still raises UniqueViolationError.
• The whole row is written on update — fields you leave unset are written with their defaults, they do not preserve the stored values (see the name field above).
• With an autoincrement primary key left unset there is nothing to conflict on, so the call degrades to a plain INSERT.
• upsert() is sugar for the primitive instance.save(on_conflict="update"), which applies insert-or-update semantics to an instance you already hold, regardless of its persistence state.

Get-or-Create¶

get_or_create(defaults={...}, **filters) looks up a row by exact-match filters and creates it when missing. It returns an (instance, created) tuple; defaults are applied only on the create path:

        customer, created = await Customer.get_or_create(
            email="[email protected]",
            defaults={"name": "Alice"},
        )
        assert created is True

        # Second call finds the existing row instead of inserting
        same, created = await Customer.get_or_create(email="[email protected]")
        assert created is False and same.id == customer.id

Update-or-Create¶

update_or_create(defaults={...}, **filters) has the same shape, but when a match exists it applies defaults to the instance and saves it:

        customer, created = await Customer.update_or_create(
            email="[email protected]",
            defaults={"plan": "pro"},
        )
        assert created is False and customer.plan == "pro"

Batch Updates¶

Update many rows in one statement — no instances are loaded — and delete the same way. update(**values) and delete() are query terminals that return the affected row count:

        # Update one instance
        post.title = "Why Ferro is *Really* Fast"
        await post.save()

        # Update many rows at once
        updated = await Post.where(lambda post: post.published == False).update(published=True)  # noqa: E712

        # Delete
        deleted = await Post.where(lambda post: post.title == "Unfinished Draft").delete()

No limit/offset on mutations¶

Portable SQL has no DELETE ... LIMIT or UPDATE ... LIMIT, so calling update() or delete() on a query with limit() or offset() set raises ValueError instead of silently mutating every matching row. To mutate a bounded subset, fetch primary keys first and mutate by primary-key set:

        free = Customer.where(lambda customer: customer.plan == "free")
        try:
            await free.limit(10).delete()
        except ValueError:
            # Portable SQL has no DELETE ... LIMIT. Fetch primary keys first,
            # then mutate by primary-key set:
            doomed = await free.limit(10).all()
            ids = [customer.id for customer in doomed]
            removed = await Customer.where(lambda customer: customer.id.in_(ids)).delete()
            assert removed == 1

Refreshing from the Database¶

refresh() reloads an instance from its primary key, discarding local state:

        # Reload an instance from the database, discarding local state
        await Customer.where(lambda customer: customer.email == "[email protected]").update(
            name="Alice L."
        )
        await customer.refresh()
        assert customer.name == "Alice L."

It raises RuntimeError if the instance has no primary key or the row no longer exists.

Deleting¶

Delete a single instance, or batch-delete via a query:

user = await User.get_or_none(42)
if user is not None:
    await user.delete()

removed = await User.where(lambda user: user.archived == True).delete()  # noqa: E712

Deleting a parent row triggers the on_delete behavior of any foreign keys pointing at it — CASCADE by default. See Delete Behavior before deleting rows with children.

Handling Database Errors¶

Every database failure raises a typed exception from the DBAPI-shaped tree rooted at FerroError — you never need to match on driver message text. Constraint violations are subclasses of IntegrityError, and the original driver detail is preserved as attributes:

        try:
            await Customer.create(email="[email protected]")  # email already taken
        except UniqueViolationError as exc:
            print(exc.sqlstate)  # "23505" (SQLSTATE) on Postgres, "2067" on SQLite
            print(exc.constraint)  # violated constraint name (Postgres only)
            print(exc.driver_message)  # original driver text, for logs

See the Exceptions API reference for the full hierarchy.

Bulk Operations and the Identity Map¶

By default Ferro keeps a per-connection identity map: loading the same primary key twice yields the same Python object, and create()/save() register instances in it.

bulk_create() is the deliberate exception — it serializes the given instances straight to the engine and skips the identity map for throughput. The instances you passed in are not registered (and auto-generated IDs are not written back onto them); re-query the rows when you need tracked instances.

inserted = await User.bulk_create([User(name="a", age=1), User(name="b", age=2)])
fresh = await User.where(lambda user: user.name.in_(["a", "b"])).all()

Not Yet Supported¶

See Also¶

• Queries — fetching and filtering data
• Exceptions — the typed error hierarchy
• Transactions — grouping mutations atomically
• Relationships — creating related records, cascade rules
• Identity Map — instance caching semantics
• Testing — testing code that mutates data