Models & CRUD¶

Introduction¶

Arvel ships with Arvent, an active-record ORM built on top of SQLAlchemy's async engine. Each database table has a corresponding "model" you use to interact with that table. Models let you query for data in your tables, as well as insert, update, and delete records.

A model is a Python class. Its instances represent rows; its class methods build queries. Because Arvent sits on SQLAlchemy, a model is a real SQLAlchemy mapped class — you get the full power of SQLAlchemy underneath the Eloquent-shaped surface.

Defining Models¶

To get started, let's create a model. The easiest way is the make:model command:

arvel make:model Flight

This generates app/models/flight.py. A model declares typed attributes using the column helpers from arvel.database:

from arvel.database import Model, Timestamps, id_, string


class Flight(Model, Timestamps):
    __tablename__ = "flights"

    id: int = id_()
    name: str = string(255)

On its own, make:model only creates the model class. To scaffold the related artifacts in one command, pass companion flags — each is named from the model:

arvel make:model Flight -mfsc        # model + migration + factory + seeder + controller
arvel make:model Flight --all        # the full set

Companions that already exist are skipped, never overwritten. Short flags combine (-mf, -mfsc).

Table Names¶

Unlike Eloquent, Arvent does not infer the table name from the class name. Every concrete model must declare __tablename__ explicitly:

class Flight(Model):
    __tablename__ = "my_flights"
    ...

Primary Keys¶

Use id_() for an auto-incrementing integer primary key, or uuid_id() for a UUID v7 key:

from arvel.database import Model, id_, uuid_id
import uuid


class Flight(Model):
    __tablename__ = "flights"
    id: int = id_()


class Booking(Model):
    __tablename__ = "bookings"
    id: uuid.UUID = uuid_id()

Both default to init=False — the value is provided by the database (integer) or a default factory (UUID), so you never pass it to the constructor. Composite primary keys are supported; find accepts a tuple in that case.

Column Helpers¶

The column helpers in arvel.database mirror the migration Blueprint vocabulary so the schema and the model speak the same language. They all return Any, so the plain annotation drives the Python type and the assignment stays clean under strict type checkers.

Most helpers accept nullable=, unique=, index=, and default=:

from decimal import Decimal
from arvel.database import Model, Timestamps, boolean, decimal, foreign_id, id_, string, text


class Post(Model, Timestamps):
    __tablename__ = "posts"

    id: int = id_()
    user_id: int = foreign_id("users.id", on_delete="CASCADE")
    title: str = string(200, index=True)
    body: str = text()
    price: Decimal = decimal(10, 2)
    is_published: bool = boolean(default=False)

Inferred Columns¶

For the common case you don't need a helper at all. The model metaclass infers a column from a bare annotation or a plain default:

class Post(Model):
    __tablename__ = "posts"
    id: int = id_()

    title: str                     # → VARCHAR(255), NOT NULL
    views: int = 0                 # → INTEGER, default 0
    published_at: datetime | None = None   # → nullable, tz-aware TIMESTAMP

When you need an option a bare annotation can't carry — a unique index, a foreign key, an explicit length — reach for field(...):

from arvel.database import field

handle: str = field(length=64, unique=True, index=True)
team_id: int = field(foreign_key="teams.id", on_delete="CASCADE")

Timestamps¶

Add the Timestamps mixin to get created_at and updated_at columns that Arvent maintains automatically — set on insert, bumped on update:

class Flight(Model, Timestamps):
    __tablename__ = "flights"
    id: int = id_()

To bump updated_at without otherwise changing the row, call await model.touch(). To run a block of work without touching timestamps:

async with Flight.without_timestamps():
    await flight.save()

Default Attribute Values¶

A column without a default becomes a required keyword argument in the model's generated constructor (models are keyword-only dataclasses). Supply a default= to make it optional:

name: str = string(255)                    # required: Flight(name=...)
slug: str | None = string(255, default=None)   # optional

Retrieving Models¶

Once you have a model and its table, you can start retrieving data. Each model serves as a query builder entry point. The all method retrieves every row:

flights = await Flight.all()

for flight in flights:
    print(flight.name)

The model's query methods (where, order_by, …) return a query builder. Add constraints, then run the query with a terminal method like get:

flights = await (
    Flight.where(active=True)
    .order_by("-name")
    .limit(10)
    .get()
)

Collections¶

all and get return a ModelCollection — a list subclass enriched with helpers like pluck, map, filter, group_by, key_by, and first_where. Because it's a list, you can iterate it, index it, and pass it anywhere a list is expected:

flights = await Flight.where(active=True).get()

names = flights.pluck("name")
grounded = flights.filter(lambda f: not f.active)

See Collections for the full method list.

Retrieving Single Models¶

In addition to retrieving all rows matching a query, you can fetch single records with find, first, and first_where:

flight = await Flight.find(1)                       # by primary key, or None
flight = await Flight.where(active=True).first()     # first match, or None
flight = await Flight.query().first_where(name="LA") # shorthand

pluck, value, and aggregates pull single columns or scalars:

names = await Flight.query().pluck("name")          # list of one column
total = await Flight.where(active=True).count()

Not Found Behavior¶

Sometimes you want to raise instead of receiving None. The *_or_fail methods raise ModelNotFoundError:

flight = await Flight.find_or_fail(1)
flight = await Flight.where(legs=3).first_or_fail()

Inserting & Updating Models¶

Inserts¶

To insert a new record, create an instance, set its attributes, and call save:

flight = Flight(name="London to Paris")
await flight.save()

Or do it in one step with create, which accepts the attributes as keyword arguments:

flight = await Flight.create(name="London to Paris")

created_at and updated_at are set automatically when the model uses Timestamps.

Updates¶

There is no instance-level update(**attrs) method. To update an existing model, set its attributes (or fill them) and call save:

flight = await Flight.find_or_fail(1)
flight.name = "Paris to London"
await flight.save()

# updates ALL matching rows in one statement
await Flight.where(active=False).update({"active": True})

To update a single model without firing model events, use update_quietly:

await flight.update_quietly(name="Paris to London")

Mass Assignment¶

create and fill go through mass-assignment protection. By default a model is unguarded — any attribute may be set. Lock it down with __fillable__ (an allowlist) or __guarded__ (a denylist):

class Flight(Model):
    __tablename__ = "flights"
    __fillable__ = ["name", "destination"]   # only these may be mass-assigned

    id: int = id_()
    name: str = string(255)
    destination: str = string(255)
    is_admin_only: bool = boolean(default=False)

A disallowed key raises MassAssignmentError. To bypass guarding deliberately, use force_fill or the unguarded context manager:

flight.force_fill(**untrusted)

with Flight.unguarded():
    flight.fill(**untrusted)

Upserts¶

Get-or-create patterns are available on the query builder:

flight = await Flight.query().first_or_create(
    {"name": "London to Paris"},          # match attributes
    {"destination": "Paris"},             # extra values if creating
)

flight = await Flight.query().update_or_create(
    {"departure": "LHR", "destination": "CDG"},
    {"price": Decimal("99.00")},
)

For bulk upserts, see Bulk Writes.

Deleting Models¶

To delete a model, call delete on an instance:

flight = await Flight.find_or_fail(1)
await flight.delete()

To delete many rows at once, run a delete through the query builder:

await Flight.where(active=False).delete()

Soft Deleting¶

Add the SoftDeletes mixin to keep rows in the table but mark them deleted with a deleted_at timestamp. Soft-deleted rows are hidden from queries by default:

from arvel.database import Model, Timestamps, SoftDeletes, id_, string


class Flight(Model, Timestamps, SoftDeletes):
    __tablename__ = "flights"
    id: int = id_()
    name: str = string(255)

Now delete sets deleted_at instead of removing the row:

await flight.delete()           # sets deleted_at; row stays
flight.trashed()                # True
await flight.restore()          # clears deleted_at
await flight.force_delete()     # permanently removes the row

Querying Soft Deleted Models¶

Soft deletes are enforced by a global scope, so ordinary queries exclude trashed rows. Include or isolate them explicitly:

await Flight.with_trashed().get()    # include soft-deleted rows
await Flight.only_trashed().get()    # only the soft-deleted rows

Bulk restore must pair with only_trashed — otherwise the default scope hides the very rows you mean to restore:

await Flight.only_trashed().restore()

Dirty Tracking¶

After loading a model you can inspect what changed before saving, and what changed after the last save:

flight = await Flight.find_or_fail(1)
flight.name = "New name"

flight.is_dirty()              # any unsaved changes?
flight.is_dirty("name")        # this attribute specifically?
flight.is_clean("destination") # unchanged?
flight.get_dirty()             # {"name": "New name"}
flight.get_original("name")    # value as loaded

await flight.save()
flight.was_changed("name")     # changed by the save just performed?
flight.get_changes()           # {attr: new_value} from the last save

Model Mixins¶

Compose behavior by inheriting mixins alongside Model:

Model Events¶

Arvent models dispatch lifecycle events you can hook into. The "before" events — creating, updating, deleting, restoring, force_deleting — can cancel the operation by returning False. The "after" and ambient events — saving, saved, created, updated, deleted, trashed, force_deleted, restored, retrieved, replicating — fire around the operation.

Observers¶

Group event handlers into an observer class. Generate one with arvel make:observer, then register it:

from arvel.database import Observer


class FlightObserver(Observer):
    def creating(self, flight: Flight) -> bool | None:
        if flight.name == "":
            return False          # cancels the insert

    async def created(self, flight: Flight) -> None:
        await notify_ops(flight)


Flight.observe(FlightObserver())

You can also register a single callback, or declare observers on the class:

Flight.on("deleted", purge_cache)


class Flight(Model, Timestamps):
    __tablename__ = "flights"
    __observed_by__ = [FlightObserver]

To dispatch model events onto the application event bus, map them with __dispatches_events__.

Muting Events¶

Suppress all events for a block of work:

async with Flight.without_events():
    await flight.save()

The *_quietly helpers (save_quietly, delete_quietly, force_delete_quietly, restore_quietly, update_quietly) do the same for a single operation.

Serializing Models¶

Models convert to dictionaries and JSON for storage or transport:

flight.to_dict()                 # respects __hidden__ / __visible__ / __appends__
flight.to_json(indent=2)
flight.to_pydantic(FlightOut)    # validate into a Pydantic schema
flight.only("id", "name")
flight.except_("internal_notes")

Control which attributes appear with the __hidden__ (denylist), __visible__ (allowlist, applied first), and __appends__ (computed accessor names) class variables, or per instance with make_hidden(...), make_visible(...), and append(...):

from arvel.database import Model, Timestamps, accessor, id_, string


class User(Model, Timestamps):
    __tablename__ = "users"
    __hidden__ = ["password_hash"]    # column dropped from output
    __appends__ = ["full_name"]       # accessor added to output

    id: int = id_()
    first_name: str = string(50)
    last_name: str = string(50)

    # full_name isn't a column — it's a computed accessor. __appends__ tells
    # to_dict()/to_json() to include it. See Casts, Accessors & Mutators.
    @accessor
    def full_name(self) -> str:
        return f"{self.first_name} {self.last_name}"

__appends__ only accepts names backed by an accessor (a @accessor method or an Attribute object) — to_dict() resolves each name with getattr(self, name), so an entry with no matching accessor raises AttributeError.

Read-Only Models¶

For models that map to a database view (or that you never want written), extend ViewModel. It blocks create, save, delete, and force_delete:

from arvel.database import ViewModel, id_, string


class MonthlyRevenue(ViewModel):
    __tablename__ = "monthly_revenue"
    id: int = id_()
    month: str = string(7)

Generate one with arvel make:model MonthlyRevenue --view (or --materialized-view).