Views API#

fastapi_restly.views implements the class-based view layer: RestView and AsyncRestView generate CRUD endpoints from a model, View is the bare primitive for hand-written endpoint groups, and include_view() registers either on a FastAPI app.

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 endpoint method: the @route, FastAPI signature, response_model, and to_response. Replace only to change the HTTP contract.
handle_<verb> — the 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 method: 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 / endpoint method. Override get_many for domain logic, handle_get_many for orchestration, to_response for the response shape; replace this method 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 “Customize RestView” in the docs):

<verb>_endpoint — the endpoint method. Owns the HTTP signature, response_model, and to_response. Rarely overridden.
handle_<verb> — the 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 method 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 / endpoint method. 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 method 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} endpoint method. Override delete for domain logic (e.g. soft delete), handle_delete for orchestration; replace this method 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 / endpoint method. Override get_many for domain logic, handle_get_many for orchestration, to_response for the response shape; replace this method 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} endpoint method. Override get_one for domain logic (visibility lives in build_query), handle_get_one for orchestration, to_response for the response shape; replace this method 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 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} endpoint method. Override update for domain logic, handle_update for orchestration, to_response for the response shape; replace this method 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] = 50#: Default page_size when the client omits it (paginated views only).

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]#

Loader options for the relationships the response schema names.

Returns recursive selectinload(...) options derived from self.schema, applied on reads (get_one / get_many) and on the post-write reload in save_object. Override to eager-load relationships the schema does not name on both paths; append to super().get_relationship_loader_options() to keep the schema-derived loads. See the “Relationship Loading and Async” how-to.

id_type#: alias of int

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]]#

paginated: ClassVar[bool] = True#: Whether list endpoints paginate. True (the default) wraps the list in a PaginatedEnvelope (data plus total_count / page / page_size / total_pages), emits page / page_size query parameters, and runs the count query. False returns every matching row in a plain Envelope (data only) with no count. For a bare array or another non-envelope shape, replace get_many_endpoint with a matching response_model (see AsyncReactAdminView).

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#

Serialize a ListingResult into the list response body.

The default shape is the {"data": [...]} envelope – with pagination metadata (total_count / page / page_size / total_pages) when paginated is true. This builds the body only; the route’s response_model is fixed from paginated separately. For a bare array or a Content-Range header, replace get_many_endpoint with a matching response_model (as AsyncReactAdminView does), not by overriding this method alone, which would fail response validation.

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

Endpoint-method 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 endpoint method.

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.Envelope(*, data: Sequence[DataT])#

Bases: BaseModel, Generic[DataT]

List response wrapper – {"data": [...]}.

The response shape for an unpaginated view. A paginated view uses PaginatedEnvelope, which adds the pagination metadata.

data: Sequence[DataT]#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

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

Bases: Generic[ModelT]

Result returned by get_many before HTTP response formatting.

total_count is None for an unpaginated listing, which does not run the count query.

objects: Sequence[ModelT]#

query_params: Any = None#

total_count: int | None = None#

class fastapi_restly.views.PaginatedEnvelope(*, data: Sequence[DataT], total_count: int, page: int, page_size: int, total_pages: int)#

Bases: Envelope

Paginated list response – data plus pagination metadata.

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

page: int#

page_size: int#

total_count: int#

total_pages: 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 / endpoint method. Override get_many for domain logic, handle_get_many for orchestration, to_response for the response shape; replace this method 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 an endpoint method asks BaseRestView.to_response() to produce.

This is separate from write-action names such as "publish". Endpoint methods 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 “Customize RestView” in the docs): the endpoint method <verb>_endpoint, the handler handle_<verb> (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 method 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 / endpoint method. 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 method only to change the HTTP contract.

delete(obj: ModelT) → None#

delete_endpoint(id: Any) → Any#: DELETE /{id} endpoint method. Override delete for domain logic (e.g. soft delete), handle_delete for orchestration; replace this method 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 / endpoint method. Override get_many for domain logic, handle_get_many for orchestration, to_response for the response shape; replace this method only to change the HTTP contract.

get_one(id: IdT) → ModelT#

get_one_endpoint(id: Any) → Any#: GET /{id} endpoint method. Override get_one for domain logic (visibility lives in build_query), handle_get_one for orchestration, to_response for the response shape; replace this method 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} endpoint method. Override update for domain logic, handle_update for orchestration, to_response for the response shape; replace this method 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[Iterable[str | Enum] | None] = 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):
    ...

Registering a view on several parents mounts its routes on each; calling include_view again with a parent the view is already registered on is a no-op.

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