Views API#

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

Class-based views with default CRUD methods and 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_action_commit -> commit -> after_action_commit). Final: call it from a custom route, never override it.
<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: scope (read visibility), authorize (policy), apply_list_params (URL grammar), to_response (wire shape), write_action (custom write actions), shared_write_action_commit (one commit over several writes). Under the verbs sit the final domain utilities (make_new_object, update_object, save_object): call them from a verb override, never override them; a server-stamped field is a column default on the model. 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_action_commit / after_action_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, 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, ResponseSchemaT, CreateSchemaT, UpdateSchemaT, IdT]

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

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

Each verb is three tiers (see “Customizing 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_action_commit -> commit -> after_action_commit); returns the domain object. Final: call it from a custom route to get the bracket, never override it.
<verb> (get_many / get_one / create / update / delete): the business method. Auth-free, commit-free; the common override point (hash a password, derive a slug, …).

async after_action_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_action_commit instead.

An exception here cannot undo the committed write, yet it fails the request, so a client that retries repeats the write. Catch and log the errors of a best-effort effect here; add an effect that must happen as an outbox row in before_action_commit.

apply_list_params(query: Select, list_params: Any) → Select#: Apply the list params (filter, sort, page) to query. Override for a non-default URL grammar; the common case is driven by configuration. The default is fastapi_restly.query.apply_list_params() with the view’s model, response class and pagination.

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 the scope, not here.

async before_action_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.

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 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 from schema_obj and save it.

Auth-free and commit-free: handle_create owns both. The usual create override point (hash a password, derive a slug), written as make_new_object(), the extra step, then save_object().

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

async delete(obj: ModelT) → None#

Remove obj and flush. Does not commit: handle_delete does.

Override (on the view or on a soft-delete mixin) to flip a timestamp instead of removing the row, without calling super(). A raw row delete elsewhere is fr.objects.async_delete_object(self.session, obj).

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

async get_many(list_params: Any, *, scope: WhereClause | Unscoped | None = None) → ListResult[ModelT]#

List the rows the scope allows, filtered and paged by list_params.

Auth-free: handle_get_many adds authorize. The query is the resolved scope (fr.resolve_scope(self), or scope when given) plus apply_list_params(). A paginated view also runs count() for total_count; a view without pagination returns every matching row with total_count=None. Relationships the response class names are eager-loaded.

The handlers always forward scope=, so an override must declare the parameter and pass it on to super(). A route’s where= arrives inside scope: handle_get_many folds it in, and there is no separate parameter to declare.

Parameters:

list_params – the list params (filter, sort, page) as the endpoint receives them.
scope – a clause that replaces the resolved scope for this read; fr.clauses.UNSCOPED reads unscoped.

async get_many_endpoint(list_params: Any) → Any#: GET / endpoint method. Override get_many for domain logic, to_response for the response shape; replace this method only to change the HTTP contract.

async get_one(id: IdT | ColumnElement[bool], *, scope: WhereClause | Unscoped | None = None) → ModelT#

Load the one row id names, through the scope, or raise 404.

Auth-free: handle_get_one adds authorize; a write action calls this directly and gates its own action. Visibility is the resolved scope (fr.resolve_scope(self), or scope when given), so a row outside it is a 404 for every caller. Relationships the response class names are eager-loaded.

The handlers always forward scope=, so an override must declare the parameter and pass it on to super().

Parameters:

id – the primary key, or a SQLAlchemy boolean expression that picks the row instead (get_one(Item.slug == slug)), which is also how a composite key is addressed. A criterion narrows inside the scope; only scope= replaces it.
scope – a clause that replaces the resolved scope for this read; fr.clauses.UNSCOPED reads unscoped.

Raises:

fastapi_restly.exc.NotFound – no row matches inside the scope. The message names an id, never a predicate.
sqlalchemy.exc.MultipleResultsFound – more than one row matches; an ambiguous key is a bug in the criterion.
TypeError – id is a Python bool (a comparison on a loaded object, which would render as WHERE true) or an uncalled clause (a scope, not a row identity).
NotImplementedError – a plain id on a composite primary key; pass a predicate.

async get_one_endpoint(id: Any) → Any#: GET /{id} endpoint method. Override get_one for domain logic (visibility lives in the scope), to_response for the response shape; replace this method only to change the HTTP contract.

async handle_create(schema_obj: CreateSchemaT) → ModelT#

Create handler: authorize, the create business method, commit bracket.

Final: override create for the domain change, authorize for the gate, and before_action_commit / after_action_commit for side effects. Call it from a custom create route, and from inside shared_write_action_commit() to share one commit with other writes; it then returns after the flush, before the commit.

async handle_delete(id: IdT | ColumnElement[bool]) → None#

Delete handler: scoped load, then delete in the commit bracket.

Final, like handle_create(): a soft delete flips a timestamp in delete, and an off-request follow-up runs in after_action_commit. Inside shared_write_action_commit(), the mutation runs immediately. The after-hook waits for the outermost block to commit.

List handler: authorize then the get_many business method.

Final, like every handler: override get_many for the query and authorize for the gate. Call it from a custom list route.

Parameters:

scope – a clause that replaces the view scope for this read, so a custom route can list another surface of the same model (a trash list); fr.clauses.UNSCOPED reads past it. Forwarded to get_many.
where – a SQLAlchemy boolean expression or a clause that narrows this read inside the scope, so a nested list keeps the visibility rules (where=Task.project_id == id). It is ANDed into the scope as fr.all_of would, and get_many receives the result as scope. The page and total_count both apply it.

Raises:

TypeError – where is a Python bool (a comparison on a loaded object), or neither an expression nor a clause.

async handle_get_one(id: IdT | ColumnElement[bool], *, scope: WhereClause | Unscoped | None = None) → ModelT#

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

Final: override get_one for the load and authorize for the gate. Call it from a custom read route as “load with scope + 404 + read-auth”. A write action instead loads with get_one(id, scope=...) and gates only its own action, the way handle_update and handle_delete do.

Parameters:

id – the primary key, or a SQLAlchemy boolean expression that picks the row instead, so a natural-key route is this handler with Item.slug == slug. Forwarded to get_one.
scope – a clause that replaces the view scope for this read, so a restore route can load the row the view scope hides. Forwarded to get_one.

async handle_update(id: IdT | ColumnElement[bool], schema_obj: UpdateSchemaT) → ModelT#

Update handler: scoped load, then update in the commit bracket.

Final, like handle_create(): update receives the loaded object, so the load, the 404, and authorize stay here. Inside shared_write_action_commit() it returns before the commit.

async make_new_object(schema_obj: CreateSchemaT) → ModelT#

Construct a new ORM object from schema_obj and add it to the session. Does not flush; save_object() does.

Final: the view-bound spelling of fr.objects.async_make_new_object, passing the view’s model. schema_obj’s own schema decides what is written, not the view’s schema. A server-stamped field (an audit id, a tenant id) is a column default on the model, which covers every write path; a value derived from the payload goes in a create override, after this call.

async save_object(obj: ModelT) → ModelT#

Flush the session and refresh obj from the database, eager-loading the relationships the response class names. Does not commit; handle_<verb> owns the commit.

Final: a side effect per write belongs in before_action_commit / after_action_commit (or a session event, to see the bulk paths too), and the reload strategy is get_relationship_loader_options.

The refresh leaves relationships unloaded, so without the eager load the serializer would reach them one lazy query at a time, which on an async session is not slow but fatal: a lazy load in the endpoint coroutine has no greenlet to suspend into and raises MissingGreenlet. Reads apply the same options in get_one / get_many.

session: _AsyncSessionDependency object at 0x7f6868f759d0>, use_cache=True, scope=function)]#

shared_write_action_commit() → AbstractAsyncContextManager[None]#

Share one commit across write actions on this session.

The outermost block commits once, then runs the queued after_action_commit hooks. write_action and the write handlers still authorize, snapshot, mutate, run before_action_commit, and flush. They return uncommitted objects inside the block:

async with self.shared_write_action_commit():
    for schema_obj in items:
        await self.handle_create(schema_obj)

Serialize returned objects and run code that depends on an after-hook only after the block exits. Nested blocks on the same session share the commit. An exception escaping a deferred-commit block aborts the shared commit, including when an enclosing deferred-commit block catches that exception. Rollback belongs to the session owner.

After-hooks run in queue order and stop on the first exception. new is the live object after all writes, while old is each action’s snapshot. Direct session.commit() calls raise RuntimeError. An async write action needs an async outermost block; a sync one joins either.

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

Apply schema_obj to the loaded obj and save it.

Auth-free and commit-free: handle_update loads obj through get_one, gates, and commits. The usual update override point, written as update_object(), the extra step, then save_object().

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

async update_object(obj: ModelT, schema_obj: UpdateSchemaT) → ModelT#

Apply writable fields from schema_obj to obj. Does not flush.

Final, like make_new_object(): an updated_by stamp is the column’s onupdate on the model; payload-derived values go in an update override, after this call.

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

Run a custom write action through the commit 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. Inside shared_write_action_commit(), the outermost block owns the commit and the after-hooks, so this bracket returns after its flush but before either one.

class fastapi_restly.views.BaseRestView#

Bases: View, Generic[ModelT, ResponseSchemaT, 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.

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

extra_query_params: ClassVar[Iterable[str]] = ()#: Extra query-parameter keys to allow on list routes beyond those derived from the response class. Use this when a view reads a custom parameter from self.request (e.g. ?verbose=true). Without this, the strict unknown-key guard rejects the request with 422. A key that the route declares, on the endpoint method or in a dependency, or an APIKeyQuery key, needs no entry.

get_relationship_loader_options() → list[Any]#

Loader options for the relationships the response class names.

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

id_type: ClassVar[type[Any] | None] = None#: The type of the {id} path parameter on the default routes. None (the default) takes the Python type of the model’s primary key, and a composite key gets int. Set a type to override it. Built-in item routes use Starlette’s int or uuid path converter for those types. Integer paths accept non-negative digits. An unmatched value falls through to another route or returns 404, instead of 422.

model: ClassVar[type[Any]]#

pagination: ClassVar[NumberedPagination | NoPagination | None] = NumberedPagination(default_page_size=50, max_page_size=1000, max_page=None, page_query_param='page', page_size_query_param='page_size', envelope=<class 'fastapi_restly._pagination.PaginatedEnvelope'>)#: How list endpoints paginate. The default NumberedPagination takes page / page_size query parameters, runs the count query, and wraps the list in its envelope (PaginatedEnvelope: data plus total_count / page / page_size / total_pages). A NoPagination returns every matching row with no count, in its envelope; None is short for NoPagination(), a plain Envelope (data only). A view inherits it, so a project base view sets it once; a view that differs changes one setting with replace() on the shared settings. For a shape no envelope model can express, such as a header, 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'}}#: OpenAPI responses documented on every route. Each class in the hierarchy adds its own, and a subclass’s entry for a status code wins.

schema: ClassVar[type[BaseModel]]#

schema_create: ClassVar[type[BaseModel]]#

schema_list_params: ClassVar[type[BaseModel]]#: The list params (filter, sort, page) as a pydantic model, generated from schema_response and model by derive_schema_list_params(). Any route method on the view that declares a list_params parameter takes it: typed for FastAPI and OpenAPI, and guarded against unknown keys like GET /, so a custom list route (a trash route naming its own scope) reads the same list params. The route can take other query parameters beside it.

schema_response: ClassVar[type[BaseModel]]#: The class of everything that goes out: the single responses, the items of the list response, and the fields the list params filter and sort on. The relationships it names are the ones the view loads. When a view sets none, Restly builds it with derive_schema_response(): the view’s schema without its WriteOnly fields, named <Resource>Response. A class that a view sets is used as it is.

schema_update: ClassVar[type[BaseModel]]#

scope: ClassVar[WhereClause | Unscoped | None] = None#: The clause every read on this view applies: list, count, and retrieve (a row outside it is 404). None (the default) falls back to the model’s declared default_scope; fr.clauses.UNSCOPED reads unscoped despite that default. Declaring a scope replaces the default, it does not stack on it; compose the replacement from the same leaves (ItemClauses.trashed containing the tenant clause visible contains). A rule that must hold under every scope is a session-level with_loader_criteria, not a view concern. See the Scopes guide.

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

to_list_response(list_result: ListResult[ModelT]) → Any#

Build the list response body: an instance of the envelope model.

The view fills its pagination’s envelope: by default a PaginatedEnvelope (data plus total_count / page / page_size / total_pages), or an Envelope (data only) for a view without pagination. The route’s response_model is the same envelope, fixed from pagination at registration. To change the shape, set the pagination’s envelope. For 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.

The page and page size come from list_result.list_params.

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

Endpoint-method response boundary.

shape selects the wire form, and with it what result is: one object for SINGLE (to_single_response()), a ListResult for LIST (to_list_response()), and nothing for EMPTY. A plain Python list is not a valid result. shape is not the write-action name. Override for envelopes or shape-wide status behavior; per-endpoint projections belong in the endpoint method.

to_single_response(obj: ModelT | ResponseSchemaT) → ResponseSchemaT#

Serialize one ORM object to the view’s schema_response.

By default the response class is the view’s schema without its WriteOnly fields, named like UserResponse. An override may return an instance of another model, such as schema. Only the fields of the response class go out. Restly copies the values of an instance of a subclass of the response class, and of schema when Restly derived the response class, without validating them again. It validates any other model into the response class from its attributes.

The view loads only the relationships that the response class names. An override that builds schema from the ORM object also reads the relationships of its WriteOnly fields. On an async view that fails with MissingGreenlet: add them in get_relationship_loader_options(), or build the response class instead.

The ORM path below validates through the response class, so a view’s 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 default envelope of NoPagination, for a view that returns every row. A paginated view uses its pagination’s envelope, PaginatedEnvelope by default.

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.ListResult(objects: Sequence[ModelT], total_count: int | None = None, list_params: Any = None)#

Bases: Generic[ModelT]

Result returned by get_many before HTTP response formatting.

total_count is None for a list without pagination, which does not run the count query. list_params holds the list params that get_many received, so BaseRestView.to_list_response() can read the page from them.

list_params: Any = None#

objects: Sequence[ModelT]#

total_count: int | None = None#

class fastapi_restly.views.NoPagination(*, envelope: type[BaseModel] = Envelope)#

Bases: object

No pagination: a list returns every matching row and runs no count query.

pagination = None is the short form of NoPagination(). Set an instance to choose the list response model:

class TagView(fr.AsyncRestView):
    pagination = fr.NoPagination(envelope=DataCount)

envelope works as for NumberedPagination, and Restly fills it by name from data and total_count, the number of rows. A generic pydantic.RootModel over the items whose before-validator returns data makes the list a bare JSON array.

replace(**changes: Any) → Self#: A copy with changes applied, checked like a new instance.

class fastapi_restly.views.NumberedPagination(*, default_page_size: int = 50, max_page_size: int = 1000, max_page: int | None = None, page_query_param: str = 'page', page_size_query_param: str = 'page_size', envelope: type[BaseModel] = PaginatedEnvelope)#

Bases: object

Page-number pagination: ?page=2&page_size=50.

Set an instance as a view’s pagination. Views inherit it, so a project base view sets it once, and a view that differs changes one setting with replace():

APP_PAGINATION = fr.NumberedPagination(
    page_size_query_param="size", max_page_size=100
)

class AppView(fr.AsyncRestView):
    pagination = APP_PAGINATION

class LogView(AppView):
    pagination = APP_PAGINATION.replace(
        default_page_size=10, max_page_size=10
    )

envelope is the list response model: a generic Pydantic model with one type parameter, which each view fills with its response schema. Restly fills its fields by name from data, total_count, page, page_size and total_pages. Leave a field out to drop it from the response, and rename one with an alias or an alias_generator. A model_validator(mode="before") receives those values as a dict and can reshape them, for example to nest the metadata. Creating the settings builds an empty page from the envelope, so a required field Restly cannot fill fails at startup instead of on a request. An envelope keeps the default extra setting: Restly passes it every page value and it keeps the fields it declares.

default_page_size: int = 50#: page_size when the client sends none.

max_page: int | None = None#: The largest page number a client may ask for, or None for no limit other than the database’s largest offset. A larger one returns 422.

max_page_size: int = 1000#: The largest page_size a client may ask for. A larger one returns 422.

page_query_param: str = 'page'#: The query parameter that carries the page number.

page_size_query_param: str = 'page_size'#: The query parameter that carries the page size.

replace(**changes: Any) → Self#

A copy with changes applied, checked like a new instance.

For a view that changes one setting of shared pagination settings:

pagination = APP_PAGINATION.replace(
    default_page_size=10, max_page_size=10
)

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.

The default envelope of NumberedPagination.

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, 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'#

LIST = 'list'#

SINGLE = 'single'#

class fastapi_restly.views.RestView#

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

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

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

Each verb is three tiers (see “Customizing 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_action_commit -> commit -> after_action_commit); returns the domain object. Final: call it from a custom route to get the bracket, never override it.
<verb> (get_many / get_one / create / update / delete): the business method. Auth-free, commit-free; the common override point (hash a password, derive a slug, …).

after_action_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_action_commit instead.

An exception here cannot undo the committed write, yet it fails the request, so a client that retries repeats the write. Catch and log the errors of a best-effort effect here; add an effect that must happen as an outbox row in before_action_commit.

apply_list_params(query: Select, list_params: Any) → Select#: Apply the list params (filter, sort, page) to query. Override for a non-default URL grammar; the common case is driven by configuration. The default is fastapi_restly.query.apply_list_params() with the view’s model, response class and pagination.

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 the scope, not here.

before_action_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.

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 query that joins a to-many relationship, whose row fan-out would otherwise inflate the count. Override for estimated counts on huge tables.

create(schema_obj: CreateSchemaT) → ModelT#

Build a new object from schema_obj and save it.

Auth-free and commit-free: handle_create owns both. The usual create override point (hash a password, derive a slug), written as make_new_object(), the extra step, then save_object().

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

delete(obj: ModelT) → None#

Remove obj and flush. Does not commit: handle_delete does.

Override (on the view or on a soft-delete mixin) to flip a timestamp instead of removing the row, without calling super(). A raw row delete elsewhere is fr.objects.delete_object(self.session, obj).

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

get_many(list_params: Any, *, scope: WhereClause | Unscoped | None = None) → ListResult[ModelT]#

List the rows the scope allows, filtered and paged by list_params.

Auth-free: handle_get_many adds authorize. The query is the resolved scope (fr.resolve_scope(self), or scope when given) plus apply_list_params(). A paginated view also runs count() for total_count; a view without pagination returns every matching row with total_count=None. Relationships the response class names are eager-loaded.

The handlers always forward scope=, so an override must declare the parameter and pass it on to super(). A route’s where= arrives inside scope: handle_get_many folds it in, and there is no separate parameter to declare.

Parameters:

list_params – the list params (filter, sort, page) as the endpoint receives them.
scope – a clause that replaces the resolved scope for this read; fr.clauses.UNSCOPED reads unscoped.

get_many_endpoint(list_params: Any) → Any#: GET / endpoint method. Override get_many for domain logic, to_response for the response shape; replace this method only to change the HTTP contract.

get_one(id: IdT | ColumnElement[bool], *, scope: WhereClause | Unscoped | None = None) → ModelT#

Load the one row id names, through the scope, or raise 404.

Auth-free: handle_get_one adds authorize; a write action calls this directly and gates its own action. Visibility is the resolved scope (fr.resolve_scope(self), or scope when given), so a row outside it is a 404 for every caller. Relationships the response class names are eager-loaded.

The handlers always forward scope=, so an override must declare the parameter and pass it on to super().

Parameters:

id – the primary key, or a SQLAlchemy boolean expression that picks the row instead (get_one(Item.slug == slug)), which is also how a composite key is addressed. A criterion narrows inside the scope; only scope= replaces it.
scope – a clause that replaces the resolved scope for this read; fr.clauses.UNSCOPED reads unscoped.

Raises:

fastapi_restly.exc.NotFound – no row matches inside the scope. The message names an id, never a predicate.
sqlalchemy.exc.MultipleResultsFound – more than one row matches; an ambiguous key is a bug in the criterion.
TypeError – id is a Python bool (a comparison on a loaded object, which would render as WHERE true) or an uncalled clause (a scope, not a row identity).
NotImplementedError – a plain id on a composite primary key; pass a predicate.

get_one_endpoint(id: Any) → Any#: GET /{id} endpoint method. Override get_one for domain logic (visibility lives in the scope), to_response for the response shape; replace this method only to change the HTTP contract.

handle_create(schema_obj: CreateSchemaT) → ModelT#

Create handler: authorize, the create business method, commit bracket.

Final: override create for the domain change, authorize for the gate, and before_action_commit / after_action_commit for side effects. Call it from a custom create route, and from inside shared_write_action_commit() to share one commit with other writes; it then returns after the flush, before the commit.

handle_delete(id: IdT | ColumnElement[bool]) → None#

Delete handler: scoped load, then delete in the commit bracket.

Final, like handle_create(): a soft delete flips a timestamp in delete, and an off-request follow-up runs in after_action_commit. Inside shared_write_action_commit(), the mutation runs immediately. The after-hook waits for the outermost block to commit.

List handler: authorize then the get_many business method.

Final, like every handler: override get_many for the query and authorize for the gate. Call it from a custom list route.

Parameters:

scope – a clause that replaces the view scope for this read, so a custom route can list another surface of the same model (a trash list); fr.clauses.UNSCOPED reads past it. Forwarded to get_many.
where – a SQLAlchemy boolean expression or a clause that narrows this read inside the scope, so a nested list keeps the visibility rules (where=Task.project_id == id). It is ANDed into the scope as fr.all_of would, and get_many receives the result as scope. The page and total_count both apply it.

Raises:

TypeError – where is a Python bool (a comparison on a loaded object), or neither an expression nor a clause.

handle_get_one(id: IdT | ColumnElement[bool], *, scope: WhereClause | Unscoped | None = None) → ModelT#

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

Final: override get_one for the load and authorize for the gate. Call it from a custom read route as “load with scope + 404 + read-auth”. A write action instead loads with get_one(id, scope=...) and gates only its own action, the way handle_update and handle_delete do.

Parameters:

id – the primary key, or a SQLAlchemy boolean expression that picks the row instead, so a natural-key route is this handler with Item.slug == slug. Forwarded to get_one.
scope – a clause that replaces the view scope for this read, so a restore route can load the row the view scope hides. Forwarded to get_one.

handle_update(id: IdT | ColumnElement[bool], schema_obj: UpdateSchemaT) → ModelT#

Update handler: scoped load, then update in the commit bracket.

Final, like handle_create(): update receives the loaded object, so the load, the 404, and authorize stay here. Inside shared_write_action_commit() it returns before the commit.

make_new_object(schema_obj: CreateSchemaT) → ModelT#

Construct a new ORM object from schema_obj and add it to the session. Does not flush; save_object() does.

Final: the view-bound spelling of fr.objects.make_new_object, passing the view’s model. schema_obj’s own schema decides what is written, not the view’s schema. A server-stamped field (an audit id, a tenant id) is a column default on the model, which covers every write path; a value derived from the payload goes in a create override, after this call.

save_object(obj: ModelT) → ModelT#

Flush the session and refresh obj from the database, eager-loading the relationships the response class names. Does not commit; handle_<verb> owns the commit.

Final: a side effect per write belongs in before_action_commit / after_action_commit (or a session event, to see the bulk paths too), and the reload strategy is get_relationship_loader_options.

The refresh leaves relationships unloaded, so without the eager load the serializer would reach them one lazy query at a time. Reads apply the same options in get_one / get_many.

session: _SyncSessionDependency object at 0x7f6868f75940>, use_cache=True, scope=function)]#

shared_write_action_commit() → AbstractContextManager[None]#

Share one commit across write actions on this session.

The outermost block commits once, then runs the queued after_action_commit hooks. write_action and the write handlers still authorize, snapshot, mutate, run before_action_commit, and flush. They return uncommitted objects inside the block:

with self.shared_write_action_commit():
    for schema_obj in items:
        self.handle_create(schema_obj)

Serialize returned objects and run code that depends on an after-hook only after the block exits. Nested blocks on the same session share the commit. An exception escaping a deferred-commit block aborts the shared commit, including when an enclosing deferred-commit block catches that exception. Rollback belongs to the session owner.

After-hooks run in queue order and stop on the first exception. new is the live object after all writes, while old is each action’s snapshot. Direct session.commit() calls raise RuntimeError. An async write action needs an async outermost block; a sync one joins either.

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

Apply schema_obj to the loaded obj and save it.

Auth-free and commit-free: handle_update loads obj through get_one, gates, and commits. The usual update override point, written as update_object(), the extra step, then save_object().

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

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

Apply writable fields from schema_obj to obj. Does not flush.

Final, like make_new_object(): an updated_by stamp is the column’s onupdate on the model; payload-derived values go in an update override, after this call.

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

Run a custom write action through the commit bracket.

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

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

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

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

Pass obj=None for writes with no single object. Exceptions skip the commit. Inside shared_write_action_commit(), the outermost block owns the commit and the after-hooks, so this bracket returns after its flush but before either one.

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() → None#: Run by include_view() once per class, before its routes are registered. A no-op here; override to adjust route methods first.

dependencies: ClassVar[Any] = None#: FastAPI dependencies run for every route, without injecting a result. Each class in the hierarchy adds its own, base first, so a subclass cannot drop a base’s guard. A subclass that lists a base’s entry again runs it where it lists it.

prefix: ClassVar[str]#: The URL prefix of every route. Each class in the hierarchy that sets one adds a segment, base first.

responses: ClassVar[dict[int | str, dict[str, Any]]] = {}#: OpenAPI responses documented on every route. Each class in the hierarchy adds its own, and a subclass’s entry for a status code wins.

route_options: ClassVar[Mapping[str | ViewRoute, Mapping[str, Any]]] = {}#: FastAPI route keyword arguments keyed by endpoint method name or ViewRoute. These override decorator metadata without replacing the endpoint method. A subclass inherits this mapping unless it declares its own, which replaces it.

tags: ClassVar[Iterable[str | Enum] | None] = None#

class fastapi_restly.views.ViewRoute(*values)#

Bases: str, Enum

Default CRUD route names that can be referenced by view options.

Values are the endpoint 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 commit 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.resolve_scope(target: type[Any] | BaseRestView[Any, Any, Any, Any, Any]) → WhereClause | Unscoped#

The scope a read applies, resolved down the ladder.

Pass a view, class or instance, for the visibility that view’s reads apply: its declared scope, the model’s default_scope when the view declares none, and fr.clauses.UNSCOPED when neither does. Pass a mapped model class for the model rung alone, which is what every reference check applies. The two answers differ wherever a view declares a scope, so code off the request path that wants what the API shows passes the view.

The result is a clause or fr.clauses.UNSCOPED, never None, so it composes without a branch:

query = fr.apply_clauses(
    select(Task).where(Task.project_id == id),
    fr.resolve_scope(TaskView),
)

Every read resolves here, so a route that applies the result sees the same rows as get_one and get_many. It resolves the scope and does not apply it: applying stays with the framework, and there is no apply-side override point.

Parameters:

target – a view class or instance, or a mapped model class.

Raises:

RestlyConfigurationError – if the declared scope is not a clause.
TypeError – if target is neither a view nor a mapped model.

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