Customize the Blog API

This tutorial extends Build a Blog API from single-method overrides through shared base classes. Its author_token and view_count demo fields are set aside. A shared base class stamps authorship server-side instead.

The examples use AsyncRestView. The same methods and patterns apply to RestView, the sync variant. Drop the async and await keywords.

The three tiers of a CRUD verb

Before overriding anything, it helps to know where each behavior lives. Every CRUD verb has three tiers, and the rule is to override the lowest tier that owns the behavior you need. The full model, including lifecycles and a decision table, is covered in Customizing RestView. From the wire inward, the tiers are:

<verb>_endpoint   the endpoint method: the @route, the FastAPI
                  signature/response_model, and to_response. Rarely overridden.
handle_<verb>     the handler: runs authorize and the commit bracket
                  (before_action_commit → commit → after_action_commit), returns the domain
                  object. Final: custom routes call it.
<verb>            the business method: the domain operation (build/apply/save).
                  Auth-free and commit-free. The usual override point.

The five verbs are get_many, get_one, create, update, and delete. So the full call chain for a create is:

POST /     → create_endpoint(schema_obj)     # endpoint method
           → handle_create(schema_obj)       # authorize + commit bracket
           → create(schema_obj)              # build + save, no commit

Three facts make this layout safe to override:

• The handler normally owns the commit. handle_<verb> runs before_action_commit, then commit, then after_action_commit around the business method.

• The business method never commits. create / update / delete build, apply, and flush. The surrounding commit bracket commits later.

• Several handlers can share one commit. Inside shared_write_action_commit(), a handler returns after its flush. The outermost block commits and runs the queued after-hooks.

Inside every method, self.session is the live database session and self.request is the FastAPI Request object.

Tier 3: the business method (the usual override point)

Most customization lives here. The business method is the domain operation: build an object, apply a payload, save it. It is auth-free and commit-free; the handler adds authorization and commit handling.

create: inject server-side fields

Real APIs rarely accept every field from the client. This example stamps the author from the current user, injected by your own auth dependency get_current_user:

from typing import Annotated
from fastapi import Depends

@fr.include_view(app)
class PostView(fr.AsyncRestView):
    prefix = "/posts"
    model = Post
    schema = PostSchema
    current_user: Annotated[User, Depends(get_current_user)]

    async def create(self, schema_obj):
        obj = await self.make_new_object(schema_obj)
        obj.author_id = self.current_user.id   # set server-side
        return await self.save_object(obj)

make_new_object builds the ORM instance. save_object flushes and refreshes it, then eager-loads the relationships the response names, but does not commit. For a field stamped on every write, see Stamping extra fields.

update: validate before saving

To reject an update based on current state, override update. It receives the loaded object:

    async def update(self, obj, schema_obj):
        if obj.published:
            raise fastapi.HTTPException(409, "Cannot edit a published post")
        obj = await self.update_object(obj, schema_obj)
        return await self.save_object(obj)

handle_update has already loaded obj through get_one and run authorize, so update only describes the domain change.

The scope: filter results to the current user

The common read customization is row visibility. get_many, count, and get_one all apply the view’s declared scope, so one clause keeps lists, totals, single-row reads, updates, and deletes aligned. Here we restrict every read to the requesting user’s own posts:

class Current(fr.ContextNamespace):
    user_id: fr.ContextParam[int]

def get_user_id(request: fastapi.Request) -> int:
    return request.state.user_id

@fr.include_view(app)
class PostView(fr.AsyncRestView):
    prefix = "/posts"
    model = Post
    schema = PostSchema
    dependencies = [Current.depends(user_id=get_user_id)]
    scope = fr.where_clause(Post.author_id == Current.user_id)

The clause and the context are declared once at module level; the generated dependency binds the per-request value. Scopes covers composing clauses and the model-wide default_scope form.

Read access has two halves, and they live in two different places:

• Visibility belongs to the scope: a hidden row is not part of this view, so get_one returns 404.

• Policy belongs to authorize, which is called in the handler. Use it for “may this caller read at all”, not for “which rows exist”.

delete: implement soft-delete

The delete business method removes the object. Override it to flip a flag instead:

from datetime import datetime, timezone


class PostView(fr.AsyncRestView):
    prefix = "/posts"
    model = Post
    schema = PostSchema

    async def delete(self, obj):
        obj.deleted_at = datetime.now(timezone.utc)
        await self.session.flush()
        # Do NOT call super(); that would remove the row.

DELETE /posts/{id} now marks the row instead of removing it. delete_endpoint still returns 204, and handle_delete still runs the commit bracket. Pair this with a scope clause that hides deleted rows. The full recipe is in Customizing RestView. The reusable mixin version is in Compose Views with Mixins.

Tier 2: the handler (authorize and the commit bracket)

One tier up from the business method sits the handler. handle_<verb> owns authorize and the commit bracket. The handler is final. Call it from a custom route to reuse that behavior. Put side-effect timing in before_action_commit or after_action_commit, and use shared_write_action_commit() when several writes need one commit. Outside a shared commit block, the defaults look like this:

handle_create  →  authorize("create", data=schema_obj)
               →  create(schema_obj)
               →  before_action_commit → commit → after_action_commit

handle_update  →  get_one(id)                     # loads through the scope
               →  authorize("update", obj, data=schema_obj)
               →  update(obj, schema_obj)
               →  before_action_commit → commit → after_action_commit

handle_delete  →  get_one(id)
               →  authorize("delete", obj)
               →  delete(obj)
               →  before_action_commit → commit → after_action_commit

The write handlers call two transaction hooks, which are the override points at this tier:

• before_action_commit(action, new, old=None) runs an in-transaction side effect (an outbox row, an audit row) that commits atomically with the write.

• after_action_commit(action, new, old=None) runs a post-commit side effect (an email, a webhook, a cache invalidation) only after the write is durable. An exception there cannot undo the write; When after_action_commit raises shows how to handle one.

Both receive old, the pre-mutation snapshot produced by snapshot(obj), so you can fire only on a real change:

    async def after_action_commit(self, action, new, old=None):
        if action == "update" and old["published"] != new.published:
            await notify_subscribers(new.id)

Stamping extra fields

The create override earlier stamped a field in the verb, which covers that verb only. A field the server owns on every write, created or updated by any view, helper, or script, is a column default on the model, reading a per-request context member:

class Current(fr.ContextNamespace):
    user_id: fr.ContextParam[int | None]


class Post(fr.TimestampsMixin, fr.IDBase):
    title: Mapped[str]
    created_by_id: Mapped[int | None] = mapped_column(
        ForeignKey("user.id"),
        default_factory=lambda: None,
        insert_default=lambda: Current.user_id(),
    )
    updated_by_id: Mapped[int | None] = mapped_column(
        ForeignKey("user.id"),
        default_factory=lambda: None,
        insert_default=lambda: Current.user_id(),
        onupdate=lambda: Current.user_id(),
    )

A Current.depends(user_id=...) entry in the view’s dependencies binds the member per request. Mark the fields fr.ReadOnly on the schema, so no payload value competes with the default. Compose Views with Mixins has the tenant and soft-delete pieces.

Object utilities

The business methods are built from a small set of object utilities that you call, never override:

create  →  make_new_object(schema_obj)   # build ORM object (no flush)
        →  save_object(obj)              # flush + refresh + eager-load (no commit)

update  →  update_object(obj, schema_obj)  # apply payload (no flush)
        →  save_object(obj)

delete  →  removes the row + flush         # no utility: override delete itself for a soft delete

make_new_object and update_object do not flush. save_object flushes, refreshes, and eager-loads the relationships the response names, but does not commit. The same operations are available as free functions for services and workers; the free save_object has no view to read a schema from, so it flushes and refreshes only.

Custom routes

Views are not limited to the default CRUD methods. Use @fr.get, @fr.post, @fr.patch, @fr.put, or @fr.delete to add endpoints. Reuse handle_get_one for a scoped load with read authorization on a read route, get_one for a scoped load only, and save_object to persist.

All route decorator keyword arguments are passed through to FastAPI, so you configure class-based routes the same way you configure regular FastAPI routes: use response_model=, status_code=, dependencies=, responses=, and the other FastAPI route options as usual.

A computed read endpoint

First we expose a summary of a post without returning the full record:

@fr.include_view(app)
class PostView(fr.AsyncRestView):
    prefix = "/posts"
    model = Post
    schema = PostSchema

    @fr.get("/{id}/summary")
    async def summary(self, id: int):
        post = await self.handle_get_one(id)   # scope + 404 + read-auth
        return {
            "id": post.id,
            "title": post.title,
            "word_count": len(post.content.split()),
        }

handle_get_one(id) gives the same scope, 404 behavior, and read authorization as GET /{id}. Use get_one(id) when you want scope and 404 without read authorization.

A state-change action

Next we add a publish action. Load with get_one, then use write_action so authorization, snapshot, commit hooks, and commit stay in the framework bracket. The bracket authorizes "publish" itself, so the load carries no read gate; the built-in write handlers load the same way:

import fastapi

    @fr.post("/{id}/publish", status_code=200)
    async def publish(self, id: int):
        post = await self.get_one(id)
        if post.published:
            raise fastapi.HTTPException(409, "Already published")
        async with self.write_action("publish", obj=post):
            post.published = True
        return self.to_response(post)

self.to_response(post) serializes through the view’s schema, the same way the inherited CRUD endpoint methods do.

If a custom action is just a create or update under another URL, call handle_create / handle_update:

    @fr.post("/{id}/repost")
    async def repost(self, id: int):
        original = await self.get_one(id)
        payload = self.schema_create(
            title=f"Repost: {original.title}",
            content=original.content,
            published=False,
        )
        return self.to_response(await self.handle_create(payload))

schema_create is the view’s create schema, generated from PostSchema when the view declares none, so the derived payload is validated like a POST / body. handle_create runs authorization, your create override, and the commit bracket.

Database conflict responses

Writes can also violate database constraints. Restly turns SQLAlchemy IntegrityError exceptions into 409 Conflict responses by default; custom envelopes and the opt-out are covered in Default Exception Handling.

Sharing behaviour with base classes

Any override above can move into a shared base class. Views are plain Python classes, so normal inheritance works.

Extract authentication into a base class

If several views need the current user, put the dependency and create-time stamp on a shared base:

from typing import Annotated
from fastapi import Depends

def get_current_user(request: fastapi.Request) -> User:
    return request.state.user   # your auth logic here


class AuthoredBase(fr.AsyncRestView):
    current_user: Annotated[User, Depends(get_current_user)]

    async def create(self, schema_obj):
        obj = await self.make_new_object(schema_obj)
        obj.author_id = self.current_user.id
        return await self.save_object(obj)


@fr.include_view(app)
class PostView(AuthoredBase):
    prefix = "/posts"
    model = Post
    schema = PostSchema


@fr.include_view(app)
class CommentView(AuthoredBase):
    prefix = "/comments"
    model = Comment
    schema = CommentSchema

FastAPI injects self.current_user on every subclass method. Register only concrete subclasses, not the base.

Extend a base-class verb with super()

A subclass can extend a base-class business method:

@fr.include_view(app)
class PostView(AuthoredBase):
    prefix = "/posts"
    model = Post
    schema = PostSchema

    async def create(self, schema_obj):
        # PostView-specific logic before the base class runs
        schema_obj.title = schema_obj.title.strip()
        return await super().create(schema_obj)

The call passes from PostView.create through AuthoredBase.create to AsyncRestView.create, and handle_create still wraps the whole chain in authorization and the commit bracket.

Apply router-level dependencies

dependencies = [Depends(fn)] applies fn to every route without injecting its result. Use it for auth guards or rate limits:

class ProtectedBase(fr.AsyncRestView):
    dependencies = [Depends(require_auth)]


@fr.include_view(app)
class PostView(ProtectedBase):
    prefix = "/posts"
    model = Post
    schema = PostSchema

Every route on /posts now runs require_auth before the endpoint function.

Share a URL namespace with prefix concatenation

When a base class defines prefix, subclass prefixes are appended: an ApiV1 base with prefix = "/api/v1" puts every subclass under /api/v1/.... The full recipe is in Share Behaviour with Base Views.

Putting it together

Here is the blog API from Build a Blog API, extended with the customizations from this tutorial. A three-line middleware stands in for real authentication so the file runs as shown:

import fastapi
import fastapi_restly as fr
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from typing import Annotated
from fastapi import Depends
from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column

fr.configure(async_database_url="sqlite+aiosqlite:///blog.db")


# --- Models ---

class Post(fr.IDBase):
    title: Mapped[str]
    content: Mapped[str]
    published: Mapped[bool] = mapped_column(default=False)
    author_id: Mapped[int | None] = mapped_column(default=None)
    deleted_at: Mapped[datetime | None] = mapped_column(default=None)


class Comment(fr.IDBase):
    content: Mapped[str]
    post_id: Mapped[int] = mapped_column(ForeignKey("post.id"))
    author_id: Mapped[int | None] = mapped_column(default=None)


@asynccontextmanager
async def lifespan(app: fastapi.FastAPI):
    # Create tables after model classes are declared so they're registered on the metadata.
    await fr.db.async_create_all(fr.IDBase)
    yield


app = fastapi.FastAPI(lifespan=lifespan)


@app.middleware("http")
async def fake_auth(request, call_next):
    request.state.user_id = 1   # demo stand-in for your real auth
    return await call_next(request)


# --- Schemas ---

class PostSchema(fr.IDSchema):
    title: str
    content: str
    published: bool


class CommentSchema(fr.IDSchema):
    content: str
    post_id: fr.MustExist[int, Post]


# --- Shared base ---

def get_current_user_id(request: fastapi.Request) -> int:
    return request.state.user_id   # set by your auth middleware


class AuthoredBase(fr.AsyncRestView):
    user_id: Annotated[int, Depends(get_current_user_id)]

    async def create(self, schema_obj):
        obj = await self.make_new_object(schema_obj)
        obj.author_id = self.user_id
        return await self.save_object(obj)


# --- Views ---

@fr.include_view(app)
class PostView(AuthoredBase):
    prefix = "/posts"
    model = Post
    schema = PostSchema

    async def update(self, obj, schema_obj):
        if obj.published:
            raise fastapi.HTTPException(409, "Cannot edit a published post")
        obj = await self.update_object(obj, schema_obj)
        return await self.save_object(obj)

    async def delete(self, obj):
        obj.deleted_at = datetime.now(timezone.utc)
        await self.session.flush()

    @fr.post("/{id}/publish", status_code=200)
    async def publish(self, id: int):
        post = await self.get_one(id)
        if post.published:
            raise fastapi.HTTPException(409, "Already published")
        async with self.write_action("publish", obj=post):
            post.published = True
        return self.to_response(post)


@fr.include_view(app)
class CommentView(AuthoredBase):
    prefix = "/comments"
    model = Comment
    schema = CommentSchema

Try it

Run the file with fastapi dev main.py, then exercise the customized behaviour:

curl -X POST http://127.0.0.1:8000/posts \
  -H 'Content-Type: application/json' \
  -d '{"title": "Hello", "content": "World", "published": false}'
# 201; the server stamps author_id on the new row

curl -X POST http://127.0.0.1:8000/posts/1/publish
# 200 with "published": true; a second call returns 409 "Already published"

curl -X PATCH http://127.0.0.1:8000/posts/1 \
  -H 'Content-Type: application/json' -d '{"title": "Edited"}'
# 409 "Cannot edit a published post"; the update override rejects it

curl -X DELETE http://127.0.0.1:8000/posts/1
# 204; the delete override sets deleted_at instead of removing the row

A follow-up GET /posts/1 still returns the post, because nothing filters soft-deleted rows yet; hiding them is the scope pairing described in the soft-delete section above.

Next steps

These pages cover the patterns from this tutorial in more detail:

• Customizing RestView: the complete override reference with all recipes

• Share Behaviour with Base Views: the full inheritance guide

• Testing: test the overrides you write

• API Reference