Tutorial Part 2: Customizing Views#

In Part 2 we extend the blog API from Part 1, working through customization from single-method overrides to shared base classes. Part 1’s author_token and view_count demo fields are set aside in this part; a shared base class stamps authorship server-side instead.

The examples use AsyncRestView. The same methods and patterns apply to RestView, the sync variant; simply drop the async/await.

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 Customize 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_commit → commit → after_commit), returns the domain
                  object. Override to change orchestration/timing.
<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

Two facts make this layout safe to override:

  • The handler owns the commit. handle_<verb> runs before_commit, then commit, then after_commit around the business method.

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

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 request context:

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

    async def create(self, schema_obj):
        obj = await self.make_new_object(schema_obj)
        obj.author_id = self.request.state.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 schema names, but does not commit. For fields stamped on both create and update, override make_new_object / update_object instead; 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.

build_query: filter results to the current user#

The common read override is row visibility. get_many, count, and get_one all use build_query, so one filter keeps listings, totals, single-row reads, updates, and deletes aligned. Here we restrict every read to the requesting user’s own posts:

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

    def build_query(self):
        user_id = self.request.state.user_id
        return super().build_query().where(Post.author_id == user_id)

Calling super().build_query() and chaining .where(...) composes cleanly with any base-class or mixin filter.

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

  • Visibility belongs to build_query: 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 = PostRead

    async def delete(self, obj):
        obj.deleted_at = datetime.now(timezone.utc)
        await self.session.flush()
        # Do NOT call super() / delete_object; 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 commits. Pair this with a build_query filter that hides deleted rows; the canonical recipe lives in Customize RestView, and the reusable mixin version in Compose Views with Mixins.

Tier 2: the handler (orchestration and timing)#

One tier up from the business method sits the handler. handle_<verb> owns authorize and the commit bracket; override it to change orchestration or timing without re-declaring the route. The defaults look like this:

handle_create  →  authorize("create", data=schema_obj)
               →  create(schema_obj)
               →  before_commit → commit → after_commit

handle_update  →  get_one(id)                     # loads through build_query
               →  authorize("update", obj, data=schema_obj)
               →  update(obj, schema_obj)
               →  before_commit → commit → after_commit

handle_delete  →  get_one(id)
               →  authorize("delete", obj)
               →  delete(obj)
               →  before_commit → commit → after_commit

The transaction hooks are the usual reason to drop to this tier:

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

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

The hooks cover most timing needs. Override handle_<verb> only when the operation order or transaction must change.

Stamping extra fields#

The create override earlier stamped a field at creation time only. For fields stamped on both create and update, override make_new_object / update_object cooperatively: call super(), mutate, and return. Base classes and mixins then compose cleanly:

    async def make_new_object(self, schema_obj):
        obj = await super().make_new_object(schema_obj)
        obj.created_by = self.request.state.user_id   # stamp the constructed object
        return obj

    async def update_object(self, obj, schema_obj):
        obj = await super().update_object(obj, schema_obj)
        obj.updated_by = self.request.state.user_id
        return obj

make_new_object builds the ORM object; update_object applies the payload. Override them for structural stamps without touching the business method.

Object utilities#

The business methods are built from a small set of object utilities. save_object and delete_object you only ever call; make_new_object and update_object you call as well, but they double as the cooperative override points from the previous section:

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

update  →  update_object(obj, schema_obj)  # apply payload (override point for stamping)
        →  save_object(obj)

delete  →  delete_object(obj)              # delete + flush (no commit)

make_new_object and update_object do not flush. save_object flushes, refreshes, and eager-loads the relationships the response schema 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 generated verbs. 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, 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 = PostRead

    @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 handle_get_one, then use write_action so authorization, snapshot, commit hooks, and commit stay in the framework bracket:

import fastapi

    @fr.post("/{id}/publish", status_code=200)
    async def publish(self, id: int):
        post = await self.handle_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 response schema, the same way the generated routes 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, schema_obj: PostRead):
        original = await self.handle_get_one(id)
        # ... derive a new payload from `original` ...
        return self.to_response(await self.handle_create(schema_obj))

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 = PostRead


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

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 = PostRead

    async def create(self, schema_obj):
        # PostView-specific logic before the base class runs
        schema_obj.slug = slugify(schema_obj.title)
        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 = PostRead

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 Part 1, extended with the customizations from this part. A three-line middleware stands in for real authentication so the file runs as is:

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 PostRead(fr.IDSchema):
    title: str
    content: str
    published: bool


class CommentRead(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 = PostRead

    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.handle_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 = CommentRead

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 build_query pairing described in the soft-delete section above.

Next steps#

The pages below go deeper into the patterns from this part: