Views API#

Class-based views: generated CRUD plus explicit override tiers.

Every CRUD verb on RestView / AsyncRestView exists at three tiers — name the tier that owns your change and override one method:

<verb>_endpoint — the route shell (wire tier): the @route, FastAPI signature, response_model, and to_response. Replace only to change the HTTP contract.
handle_<verb> — the request handler: runs authorize and the commit bracket (before_commit -> commit -> after_commit). Override for orchestration or timing.
<verb> (get_many, get_one, create, update, delete) — the business verb: the domain operation, auth-free and commit-free. The usual override point.

Cross-cutting seams: build_query (read scope/visibility), authorize (policy), apply_query_params (URL grammar), to_response (wire shape), write_action (custom write brackets). View is the bare class-based primitive for non-CRUD endpoint groups (auth flows, webhooks, RPC).

class fastapi_restly.views.Action#

Bases: object

Canonical CRUD action names passed to authorize / before_commit / after_commit.

This is a constants class, not an Enum: custom actions and mixins add their own names. Use constants for typo checking at import time.

CREATE = 'create'#

DELETE = 'delete'#

GET_MANY = 'get_many'#

GET_ONE = 'get_one'#

UPDATE = 'update'#

class fastapi_restly.views.AsyncReactAdminView#

Bases: _ReactAdminMixin, AsyncRestView

AsyncRestView that speaks the ra-data-simple-rest wire contract.

Use this instead of AsyncRestView when your frontend is react-admin with ra-data-simple-rest.

async get_many_endpoint() → Any#: GET / route shell (wire tier). Override get_many for domain logic, handle_get_many for orchestration, to_response for the response shape; replace this shell only to change the HTTP contract.

async put(id: Any, schema_obj: Any) → Any#

class fastapi_restly.views.AsyncRestView#

Bases: BaseRestView[ModelT, SchemaT, CreateSchemaT, UpdateSchemaT, IdT]

AsyncRestView creates an async CRUD/REST interface for database objects. Basic usage:

class FooView(AsyncRestView):
    prefix = "/foo"
    schema = FooRead
    model = Foo

Each verb is three tiers (see “the handle design” in the docs):

<verb>_endpoint — the route shell (wire). Owns the HTTP signature, response_model, and to_response. Rarely overridden.
handle_<verb> — the request handler. Owns authorize and the commit bracket (before_commit -> commit -> after_commit); returns the domain object. Reuse from custom actions to get the bracket.
<verb> (get_many / get_one / create / update / delete) — the domain operation. Auth-free, commit-free; the common override point (hash a password, derive a slug, …).

async after_commit(action: str, new: ModelT | None, old: dict[str, Any] | None = None) → None#

Post-commit side effect (email, webhook, cache invalidation). old enables dirty detection (“notify only if the status changed”).

For external effects only: the write is already durable, so mutating new or the database here is NOT persisted. A mutation to new also leaks into this request’s response (which serializes new after this hook) while being silently discarded from storage – do the mutation in the business verb or before_commit instead.

apply_query_params(query: Select, query_params: Any) → Select#: Apply URL filter/sort/pagination to query. Override for a non-default URL grammar; the common case is driven by configuration.

async authorize(action: str, obj: ModelT | None = None, data: Any = None) → None#

Gate a verb. Called by handle_<verb> at the right phase: before the write for create, and after the scoped load for update / delete / get_one (so obj is available for row-level checks).

The default is a no-op – override to enforce policy, raising fr.exc.Forbidden / fr.exc.NotFound to reject (action says which verb; obj / data carry the loaded row and the request payload). Row visibility – hiding a row from every caller – belongs in build_query, not here.

async before_commit(action: str, new: ModelT | None, old: dict[str, Any] | None = None) → None#: In-transaction side effect (outbox rows, audit rows), committed atomically with the write. old is the pre-mutation snapshot dict.

build_query() → Select#

Return the base SQLAlchemy Select used by every read on this view’s model – list, count, and retrieve. Override to add WHERE clauses that should apply to all of them (tenant scope, soft-delete filtering, row-level permission visibility). Call super().build_query() and chain .where(...) to compose with base-class or mixin filters.

Retrieve also routes through this query, so a row hidden from the list returns 404 from GET /{id}.

async count(query: Select) → int#

Total for the list, ignoring presentation-layer ordering/pagination.

The stripped query is made DISTINCT and wrapped as a subquery, so the total is correct across user-provided query shapes – including a build_query that joins a to-many relationship, whose row fan-out would otherwise inflate the count. Override for estimated counts on huge tables.

async create(schema_obj: CreateSchemaT) → ModelT#: Build a new object and save it. Override from scratch for domain logic (e.g. hash a password): never commits, so the bracket can’t break.

async create_endpoint(schema_obj: Any) → Any#: POST / route shell (wire tier). Override create for domain logic (it is commit-free; the handler owns the commit), handle_create for orchestration, to_response for the response shape; replace this shell only to change the HTTP contract.

async delete(obj: ModelT) → None#: Delete obj. Override (e.g. on a soft-delete mixin) to flip a timestamp instead of removing the row.

async delete_endpoint(id: Any) → Any#: DELETE /{id} route shell (wire tier). Override delete for domain logic (e.g. soft delete), handle_delete for orchestration; replace this shell only to change the HTTP contract (e.g. return the deleted object instead of 204).

async get_many(query_params: Any) → ListingResult[ModelT]#

Return the scoped, filtered, paginated page plus the total count.

Routes through build_query() (scope) + apply_query_params() (filter/sort/page) + count(). Auth-free; handle_get_many adds the authorize call.

async get_many_endpoint(query_params: Any) → Any#: GET / route shell (wire tier). Override get_many for domain logic, handle_get_many for orchestration, to_response for the response shape; replace this shell only to change the HTTP contract.

async get_one(id: IdT) → ModelT#

Load one object through build_query() (scope + 404).

Auth-free: visibility comes from build_query, so a row hidden by the scope is a clean 404 for every caller. handle_get_one adds read-auth.

async get_one_endpoint(id: Any) → Any#: GET /{id} route shell (wire tier). Override get_one for domain logic (visibility lives in build_query), handle_get_one for orchestration, to_response for the response shape; replace this shell only to change the HTTP contract.

async handle_create(schema_obj: CreateSchemaT) → ModelT#

async handle_delete(id: IdT) → None#

async handle_get_many(query_params: Any) → ListingResult[ModelT]#: List request handler: authorize then the get_many domain op.

async handle_get_one(id: IdT) → ModelT#

Retrieve handler: scoped load (404 by visibility) then read-auth.

Reusable from custom actions as “load with scope + 404 + read-auth”.

async handle_update(id: IdT, schema_obj: UpdateSchemaT) → ModelT#

session: Annotated[AsyncSession, Depends(dependency=_async_generate_session, use_cache=True, scope=function)]#

async update(obj: ModelT, schema_obj: UpdateSchemaT) → ModelT#: Apply the update payload to obj and save it.

async update_endpoint(id: Any, schema_obj: Any) → Any#: PATCH /{id} route shell (wire tier). Override update for domain logic, handle_update for orchestration, to_response for the response shape; replace this shell only to change the HTTP contract.

write_action(action: str, *, obj: ~typing.Any = <object object>, data: ~typing.Any = None)#

Run a custom write action through the standard write bracket.

Use this for non-CRUD actions such as publish or change-password:

async with self.write_action("publish", obj=article):  # in-place
    article.status = "published"

For create-shaped actions, omit obj and set w.obj before exit:

async with self.write_action("create", data=req) as w:
    w.obj = await self.make_new_object(req)

Pass obj=None for writes with no single object. Exceptions skip the commit.

class fastapi_restly.views.BaseRestView#

Bases: View, Generic[ModelT, SchemaT, CreateSchemaT, UpdateSchemaT, IdT]

Base class for RestView implementations.

This class contains the common functionality shared between AsyncRestView and RestView, including schema definitions, model configuration, and common CRUD operation logic.

classmethod before_include_view()#

Apply type annotations needed for FastAPI, before creating an APIRouter from this view and registering it.

This function can be overridden to further tweak the endpoints before they are added to FastAPI.

default_page_size: ClassVar[int | None] = None#: Default page_size for list endpoints. None means “no implicit cap” (the framework default). Override per-view.

exclude_routes: ClassVar[Iterable[str | ViewRoute]] = ()#

extra_query_params: ClassVar[Iterable[str]] = ()#: Extra query-parameter keys to allow on the listing endpoint beyond those derived from the response schema. Use this when a view consumes a custom parameter (e.g. ?include_deleted=true on a soft-delete mixin). Without this, the strict unknown-key guard rejects the request with 422.

get_relationship_loader_options() → list[Any]#

id_type#: alias of int

include_pagination_metadata: ClassVar[bool] = False#

listing_param_schema: ClassVar[type[BaseModel]]#

max_page_size: ClassVar[int] = 1000#: Maximum page_size accepted on list endpoints. Above this returns 422.

model: ClassVar[type[DeclarativeBase]]#

pagination_response_schema: ClassVar[type[BaseModel]]#

request: Request#

responses: ClassVar[dict[int | str, dict[str, Any]]] = {404: {'description': 'Not found'}}#

schema: ClassVar[type[BaseModel]]#

schema_create: ClassVar[type[BaseModel]]#

schema_update: ClassVar[type[BaseModel]]#

snapshot(obj: Any) → dict[str, Any]#: Frozen capture of an object’s already-loaded column values, passed as old to before_commit / after_commit for dirty detection. Override to change what old captures (e.g. include a relationship’s prior state); the default delegates to fastapi_restly.snapshot().

to_listing_response(query_params: Any, listing_result: ListingResult[ModelT]) → Any#

to_paginated_listing_response(query_params: Any, listing_result: ListingResult[Any]) → dict[str, Any]#

to_response(obj_or_list: Any, shape: ResponseShape = ResponseShape.SINGLE) → Any#

Route-shell response boundary.

shape selects the wire form: single object, listing, or empty. It is not the write-action name. Override for envelopes or shape-wide status behavior; per-endpoint projections belong in the route shell.

to_response_schema(obj: ModelT | SchemaT) → SchemaT#

Serialize an ORM object to the configured response schema.

WriteOnly fields are stripped from responses by exclude=True on the marker itself (recursively, at serialization time), so a pre-built schema instance is safe to return as-is. The ORM path below still validates through the WriteOnly-omitting response schema, so a read schema that declares a WriteOnly field the ORM object doesn’t carry (e.g. password backed by a password_hash column) doesn’t fail response validation.

class fastapi_restly.views.ListingResult(objects: Sequence[ModelT], total_count: int, query_params: Any = None)#

Bases: Generic[ModelT]

Result returned by get_many before HTTP response formatting.

objects: Sequence[ModelT]#

query_params: Any = None#

total_count: int#

class fastapi_restly.views.ReactAdminView#

Bases: _ReactAdminMixin, RestView

RestView that speaks the ra-data-simple-rest wire contract.

Use this instead of RestView when your frontend is react-admin with ra-data-simple-rest.

get_many_endpoint() → Any#: GET / route shell (wire tier). Override get_many for domain logic, handle_get_many for orchestration, to_response for the response shape; replace this shell only to change the HTTP contract.

put(id: Any, schema_obj: Any) → Any#

class fastapi_restly.views.ResponseShape(*values)#

Bases: str, Enum

The wire shape a route shell asks BaseRestView.to_response() to produce.

This is separate from write-action names such as "publish". Route shells choose one of these three response shapes; custom actions remain an open string namespace.

EMPTY = 'empty'#

LISTING = 'listing'#

SINGLE = 'single'#

class fastapi_restly.views.RestView#

Bases: BaseRestView[ModelT, SchemaT, CreateSchemaT, UpdateSchemaT, IdT]

RestView creates a sync CRUD/REST interface for database objects. Basic usage:

class FooView(RestView):
    prefix = "/foo"
    schema = FooRead
    model = Foo

Each verb is three tiers (see “the handle design” in the docs): the <verb>_endpoint route shell, the handle_<verb> request handler (authorize + commit bracket), and the bare verb <verb> (the domain operation – the common override point).

after_commit(action: str, new: ModelT | None, old: dict[str, Any] | None = None) → None#

Post-commit side effect (email, webhook, cache).

For external effects only: the write is already durable, so mutating new or the database here is NOT persisted (and a mutation to new leaks into this request’s response while being discarded from storage). Do the mutation in the business verb or before_commit instead.

apply_query_params(query: Select, query_params: Any) → Select#: Apply URL filter/sort/pagination to query.

authorize(action: str, obj: ModelT | None = None, data: Any = None) → None#: Gate a verb. Sync counterpart of AsyncRestView.authorize() – a no-op by default; override to enforce policy and raise fr.exc.Forbidden / fr.exc.NotFound to reject. Row visibility belongs in build_query.

before_commit(action: str, new: ModelT | None, old: dict[str, Any] | None = None) → None#: In-transaction side effect (outbox/audit), atomic with the write.

build_query() → Select#: Return the base SQLAlchemy Select used by every read – list, count, and retrieve. Override to add WHERE clauses (tenant scope, soft-delete, row-level visibility) that apply to all three.

count(query: Select) → int#

Total for the list, ignoring presentation ordering/pagination.

Made DISTINCT before counting so a build_query that joins a to-many relationship doesn’t inflate the total via row fan-out.

create(schema_obj: CreateSchemaT) → ModelT#

create_endpoint(schema_obj: Any) → Any#: POST / route shell (wire tier). Override create for domain logic (it is commit-free; the handler owns the commit), handle_create for orchestration, to_response for the response shape; replace this shell only to change the HTTP contract.

delete(obj: ModelT) → None#

delete_endpoint(id: Any) → Any#: DELETE /{id} route shell (wire tier). Override delete for domain logic (e.g. soft delete), handle_delete for orchestration; replace this shell only to change the HTTP contract (e.g. return the deleted object instead of 204).

get_many(query_params: Any) → ListingResult[ModelT]#

get_many_endpoint(query_params: Any) → Any#: GET / route shell (wire tier). Override get_many for domain logic, handle_get_many for orchestration, to_response for the response shape; replace this shell only to change the HTTP contract.

get_one(id: IdT) → ModelT#

get_one_endpoint(id: Any) → Any#: GET /{id} route shell (wire tier). Override get_one for domain logic (visibility lives in build_query), handle_get_one for orchestration, to_response for the response shape; replace this shell only to change the HTTP contract.

handle_create(schema_obj: CreateSchemaT) → ModelT#

handle_delete(id: IdT) → None#

handle_get_many(query_params: Any) → ListingResult[ModelT]#

handle_get_one(id: IdT) → ModelT#

handle_update(id: IdT, schema_obj: UpdateSchemaT) → ModelT#

session: Annotated[Session, Depends(dependency=_generate_session, use_cache=True, scope=function)]#

update(obj: ModelT, schema_obj: UpdateSchemaT) → ModelT#

update_endpoint(id: Any, schema_obj: Any) → Any#: PATCH /{id} route shell (wire tier). Override update for domain logic, handle_update for orchestration, to_response for the response shape; replace this shell only to change the HTTP contract.

write_action(action: str, *, obj: ~typing.Any = <object object>, data: ~typing.Any = None)#

Run a custom write action through the standard write bracket.

Use this for non-CRUD actions such as publish or change-password:

with self.write_action("publish", obj=article):
    article.status = "published"

For create-shaped actions, omit obj and set w.obj before exit. Pass obj=None for writes with no single object. Exceptions skip the commit.

class fastapi_restly.views.View#

Bases: object

Class-based view primitive for FastAPI.

Group related endpoints on a class, share dependencies and metadata via class attributes, and let subclasses override individual handlers. Routes are bound at include_view() time, not at class-definition time, so subclassing works the way Python developers expect: override a method on a subclass and the override is what runs.

Most users will subclass RestView or AsyncRestView, which extend View with CRUD scaffolding. Use View directly for grouped non-CRUD endpoints (auth flows, custom RPC routes, etc.).

classmethod before_include_view()#

dependencies: ClassVar[Any] = None#

prefix: ClassVar[str]#

responses: ClassVar[dict[int | str, dict[str, Any]]] = {}#

tags: ClassVar[Any] = None#

class fastapi_restly.views.ViewRoute(*values)#

Bases: str, Enum

Generated CRUD routes that can be referenced by view options.

Values are the route-shell method names so exclude_routes can drop them.

CREATE = 'create_endpoint'#

DELETE = 'delete_endpoint'#

GET_MANY = 'get_many_endpoint'#

GET_ONE = 'get_one_endpoint'#

UPDATE = 'update_endpoint'#

async fastapi_restly.views.async_run_write_action(host: AsyncWriteHost, action: str, *, obj: Any = None, data: Any = None, mutate: Callable[[], Awaitable[T]]) → T#: Run mutate inside the async write bracket and return its result.

fastapi_restly.views.delete(path: str, **api_route_kwargs: Any) → Callable[[...], Any]#

Decorator to mark a View method as a DELETE endpoint.

Equivalent to:

@route(path, methods=["DELETE"], status_code=204, ... )

fastapi_restly.views.get(path: str, **api_route_kwargs: Any) → Callable[[...], Any]#

Decorator to mark a View method as a GET endpoint.

Equivalent to:

@route(path, methods=["GET"], status_code=200, ... )

fastapi_restly.views.include_view(parent_router: APIRouter | FastAPI, view_cls: V | None = None) → V | Callable[[V], V]#

Add a View class’s routes to a FastAPI app or APIRouter.

Prefer the direct call form from your app/router composition layer:

include_view(app, MyView)

For small apps, it can also be used as a decorator:

@include_view(app)
class MyView(AsyncRestView):
    ...

fastapi_restly.views.patch(path: str, **api_route_kwargs: Any) → Callable[[...], Any]#

Decorator to mark a View method as a PATCH endpoint.

Equivalent to:

@route(path, methods=["PATCH"], status_code=200, ... )

fastapi_restly.views.post(path: str, **api_route_kwargs: Any) → Callable[[...], Any]#

Decorator to mark a View method as a POST endpoint.

Equivalent to:

@route(path, methods=["POST"], status_code=201, ... )

fastapi_restly.views.put(path: str, **api_route_kwargs: Any) → Callable[[...], Any]#

Decorator to mark a View method as a PUT endpoint.

Equivalent to:

@route(path, methods=["PUT"], status_code=200, ... )

fastapi_restly.views.route(path: str, **api_route_kwargs: Any) → Callable[[...], Any]#

Decorator to mark a View method as an endpoint. The path and api_route_kwargs are passed into APIRouter.add_api_route(), see for example: https://fastapi.tiangolo.com/reference/apirouter/#fastapi.APIRouter.get

Endpoints methods are later added as routes to the FastAPI app using include_view()

fastapi_restly.views.run_write_action(host: WriteHost, action: str, *, obj: Any = None, data: Any = None, mutate: Callable[[], T]) → T#: Sync variant of async_run_write_action().