Quickstart¶

In about 10 minutes you'll build a small blog backend: two related models (Author and Post), an in-memory SQLite database, and every core Ferro operation — create, query, traverse relationships, update, delete, and transactions.

Every code block on this page comes from one runnable script, shown in full at the bottom of the page. Follow along in a file of your own, or just run the script.

Define Your Models¶

Ferro supports two equivalent field-declaration styles — options on the assignment side, or inside typing.Annotated. Every model example in these docs shows both; pick one and stay consistent in your project.

AssignmentAnnotated

from datetime import datetime
from typing import Annotated

from ferro import BackRef, Field, ForeignKey, Model, Relation, connect, engines, transaction


class Author(Model):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    email: str = Field(unique=True)
    posts: Relation[list["Post"]] = BackRef()


class Post(Model):
    id: int | None = Field(default=None, primary_key=True)
    title: str
    body: str
    published: bool = False
    created_at: datetime = Field(default_factory=datetime.now)
    author: Annotated[Author, ForeignKey(related_name="posts")]

from datetime import datetime
from typing import Annotated

from ferro import BackRef, Field, ForeignKey, Model, Relation, connect, engines


class Author(Model):
    id: Annotated[int | None, Field(default=None, primary_key=True)]
    name: str
    email: Annotated[str, Field(unique=True)]
    posts: Relation[list["Post"]] = BackRef()


class Post(Model):
    id: Annotated[int | None, Field(default=None, primary_key=True)]
    title: str
    body: str
    published: bool = False
    created_at: Annotated[datetime, Field(default_factory=datetime.now)]
    author: Annotated[Author, ForeignKey(related_name="posts")]

A Ferro model is a Pydantic model — annotated fields become columns, and defaults work exactly as in Pydantic:

Field(default=None, primary_key=True) marks id as the primary key. It's int | None because the database assigns it on insert.
Field(unique=True) adds a unique constraint; default_factory=datetime.now gives each post a creation timestamp.
Annotated[Author, ForeignKey(related_name="posts")] declares the many-to-one side: each Post stores an author_id column pointing at an Author.
Relation[list["Post"]] = BackRef() is the reverse side: author.posts becomes a chainable query for that author's posts. related_name="posts" is what links the two.

Connect¶

    await connect("sqlite::memory:", auto_migrate=True)

connect() takes a database URL. sqlite::memory: gives you a throwaway in-memory database — perfect for this tutorial and for tests. For a file-backed database use sqlite:app.db?mode=rwc (rwc = read/write/create), or a postgres://... URL for PostgreSQL.

auto_migrate=True creates tables for every registered model on connect. It's great for development; for production schemas, use Alembic migrations.

Create Data¶

        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),
            ]
        )

Model.create(...) validates the data, inserts one row, and returns the instance with its database-assigned id populated. Notice you can pass a model instance (author=alice) for the foreign key. It never overwrites an existing row — a duplicate raises UniqueViolationError.
Model.bulk_create([...]) inserts many rows in a single statement — use it whenever you're loading more than a handful of rows. Here we set author_id directly instead of passing the instance.

Query¶

        # Fetch by primary key
        same_post = await Post.get(post.id)

        # Filter, order, and slice
        published = (
            await Post.where(lambda post: post.published == True)  # noqa: E712
            .order_by(lambda post: post.created_at, "desc")
            .limit(10)
            .all()
        )

        # Aggregate terminals
        total = await Post.select().count()
        has_drafts = await Post.where(lambda post: post.published == False).exists()  # noqa: E712

Post.get(pk) fetches one row by primary key.
where(...) filters, order_by(...) sorts, limit(...) slices — and nothing touches the database until a terminal like .all(), .first(), .count(), or .exists() runs the query.
lambda post: post.published == True is a lambda predicate — the only predicate style where() accepts. Name the parameter after the row type in lowercase singular (user for User, post for Post). See Queries for the full operator surface and how column names are validated at build time.

What happened

Thanks to Ferro's identity map, Post.get(post.id) returns the same Python object as the post you created earlier — not a duplicate copy. One row, one instance.

Work with Relationships¶

        # Forward access: awaiting the foreign key loads the related instance
        author = await same_post.author

        # Reverse access: the BackRef is a chainable query
        alice_posts = await author.posts.where(lambda post: post.published == True).all()  # noqa: E712

Two directions, two idioms:

Forward (post.author): awaiting the foreign key attribute loads the related Author.
Reverse (author.posts): the BackRef is a query, so you can chain .where(), .order_by(), and friends before awaiting it.

Update & Delete¶

        # 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()

For a single instance: mutate attributes, then await post.save(). save() UPDATEs an instance that came from the database and INSERTs one that never did — see Mutations.
For many rows: chain .update(field=value) or .delete() onto a where() query. Both return the number of affected rows.

Wrap It in a Transaction¶

        async with transaction():
            bob = await Author.create(name="Bob", email="[email protected]")
            await Post.create(title="Hello", body="...", author=bob)
        # Commits on success, rolls back if the block raises

Everything inside async with transaction(): commits together when the block exits cleanly — and rolls back entirely if it raises. Use it whenever multiple writes must succeed or fail as one. More in Transactions.

Complete Script¶

The whole tutorial as one runnable file — it lives in the repo at docs/examples/quickstart.py:

"""Runnable companion to the Quickstart tutorial (docs/pages/getting-started/quickstart.md)."""

import asyncio

from datetime import datetime
from typing import Annotated

from ferro import BackRef, Field, ForeignKey, Model, Relation, connect, engines, transaction


class Author(Model):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    email: str = Field(unique=True)
    posts: Relation[list["Post"]] = BackRef()


class Post(Model):
    id: int | None = Field(default=None, primary_key=True)
    title: str
    body: str
    published: bool = False
    created_at: datetime = Field(default_factory=datetime.now)
    author: Annotated[Author, ForeignKey(related_name="posts")]


async def main() -> None:
    await connect("sqlite::memory:", auto_migrate=True)

    async with engines.session():
        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),
            ]
        )
        assert post.id is not None

        # Fetch by primary key
        same_post = await Post.get(post.id)

        # Filter, order, and slice
        published = (
            await Post.where(lambda post: post.published == True)  # noqa: E712
            .order_by(lambda post: post.created_at, "desc")
            .limit(10)
            .all()
        )

        # Aggregate terminals
        total = await Post.select().count()
        has_drafts = await Post.where(lambda post: post.published == False).exists()  # noqa: E712
        assert same_post is post  # identity map: same Python object
        assert len(published) == 2
        assert total == 3
        assert has_drafts

        # Forward access: awaiting the foreign key loads the related instance
        author = await same_post.author

        # Reverse access: the BackRef is a chainable query
        alice_posts = await author.posts.where(lambda post: post.published == True).all()  # noqa: E712
        assert author.name == "Alice"
        assert len(alice_posts) == 2

        # 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()
        assert updated == 1
        assert deleted == 1

        async with transaction():
            bob = await Author.create(name="Bob", email="[email protected]")
            await Post.create(title="Hello", body="...", author=bob)
        # Commits on success, rolls back if the block raises
        assert await Author.select().count() == 2

    print("quickstart example ran successfully")


if __name__ == "__main__":
    asyncio.run(main())

What's Next¶

Next Steps — pick a path based on what you're building
Models & Fields — every field type and constraint
Queries — the lambda predicate style in depth, ordering, slicing, terminals
Relationships — foreign keys, back-references, many-to-many