Skip to content

API Views

Core view base classes. Import these directly from fastapi_views.views or from fastapi_views.views.api.

View is the lowest level — route registration plus JSON serialization driven by the endpoint's return annotation. APIView adds response_schema, per-action dependencies, response-header documentation and error handling. The *APIView mixin classes each implement a single CRUD action and can be combined freely.

For a complete walkthrough see Basic usage.


View

Bases: DependencyMixin, ABC

Base View Class

Source code in fastapi_views/views/api.py
class View(DependencyMixin, ABC):
    """Base View Class"""

    api_component_name: str
    errors: tuple[type[APIError], ...] = ()
    from_attributes: bool | None = None
    validate_response: bool = True
    _serializers: ClassVar[TypeAdapterMap] = {}

    def __init__(self, request: Request, response: Response) -> None:
        self.request = request
        self.response = response

    def _set_default_media_type(self, media_type: str) -> None:
        if self.response.media_type is None:
            self.response.media_type = media_type

    @classmethod
    def get_name(cls) -> str:
        return getattr(cls, "api_component_name", cls.__name__)

    @classmethod
    def get_slug_name(cls) -> str:
        return f"{cls.get_name().lower().replace(' ', '_')}"

    def get_response(
        self,
        content: Any,
        *,
        status_code: int = HTTP_200_OK,
        schema: Any = None,
        headers: dict[str, str] | None = None,
    ) -> Response:
        if isinstance(content, Response):
            return self.finalize_response(content)

        self.response.status_code = status_code

        if content is not None and not isinstance(content, (str, bytes)):
            serializer = self.get_serializer(schema)
            content = self.get_json_content(content=content, serializer=serializer)
            self._set_default_media_type("application/json")

        if isinstance(content, str):
            content = content.encode(self.response.charset)
            self._set_default_media_type("text/plain")
        if isinstance(content, bytes):
            self.response.body = content

        # Headers may already be set on the response (e.g. inside the view).
        # ``init_headers`` rebuilds ``raw_headers`` from ``headers`` alone, so
        # preserve the pre-existing ones it does not regenerate.
        preset = list(self.response.raw_headers)
        self.response.init_headers(headers)
        generated = {key for key, _ in self.response.raw_headers}
        self.response.raw_headers[:0] = [
            item for item in preset if item[0] not in generated
        ]
        # ``init_headers`` swapped in a fresh ``raw_headers`` list. FastAPI cached
        # a ``MutableHeaders`` over the *previous* list when it built the response
        # (``del response.headers["content-length"]``), so drop that stale cache
        # to keep ``response.headers`` in sync with what is actually sent.
        self.response.__dict__.pop("_headers", None)
        return self.finalize_response(self.response)

    def finalize_response(self, response: Response) -> Response:
        """Hook to post-process the built response before it is returned.

        Returns it unchanged by default; mixins such as
        :class:`~fastapi_views.views.mixins.ConditionalMixin` override this to
        attach validators and downgrade to ``304``.
        """
        return response

    def get_serializer(self, schema: Any | None) -> TypeAdapter[Any]:
        if schema is None:
            return AnyTypeAdapter
        if schema not in self._serializers:
            self._serializers[schema] = TypeAdapter(schema)
        return self._serializers[schema]

    def get_json_content(self, content: Any, serializer: TypeAdapter[Any]) -> Any:
        if self.validate_response:
            content = serializer.validate_python(
                content,
                from_attributes=self.from_attributes,
            )
        return serializer.dump_json(content)

    @classmethod
    def get_api_actions(cls, prefix: str = "") -> Generator[dict[str, Any], Any, None]:
        yield from cls.get_custom_api_actions(prefix)

    @classmethod
    def get_custom_endpoint(
        cls,
        func: Callable[Concatenate[View, P], Any],
    ) -> Callable[Concatenate[View, P], Any]:
        options = getattr(func, "kwargs", {})
        status_code = options.get("status_code", HTTP_200_OK)
        schema = options.get("response_model", get_type_hints(func).get("return"))

        async def _async_endpoint(
            self: View,
            *args: P.args,
            **kwargs: P.kwargs,
        ) -> Response:
            res = await func(self, *args, **kwargs)
            return self.get_response(res, status_code=status_code, schema=schema)

        def _sync_endpoint(self: View, *args: P.args, **kwargs: P.kwargs) -> Response:
            res = func(self, *args, **kwargs)
            return self.get_response(res, status_code=status_code, schema=schema)

        endpoint = (
            _async_endpoint if inspect.iscoroutinefunction(func) else _sync_endpoint
        )

        cls._patch_endpoint_signature(endpoint, func)
        return endpoint

    @classmethod
    def _is_endpoint(cls, member: Any) -> bool:
        return callable(member) and hasattr(member, VIEWSET_ROUTE_FLAG)

    @staticmethod
    def _is_response_model(annotation: Any) -> bool:
        """Whether a return annotation is usable as an OpenAPI response model."""
        if annotation is None or annotation is type(None):
            return False
        return not _contains_response_type(annotation)

    @classmethod
    def get_custom_api_actions(
        cls,
        prefix: str = "",
    ) -> Generator[dict[str, Any], None, None]:
        for _, route_endpoint in inspect.getmembers(cls, cls._is_endpoint):
            endpoint = cls.get_custom_endpoint(route_endpoint)
            options = getattr(route_endpoint, "kwargs", {})
            route_prefix = prefix
            if options.get("detail"):
                route_prefix += cls.get_action_detail_route()
            extra: dict[str, Any] = {}
            # Document what the endpoint actually serializes: the runtime
            # serializer falls back to the return annotation, so OpenAPI must
            # prefer it too (before the view-level response_schema default).
            if "response_model" not in options:
                return_annotation = get_type_hints(route_endpoint).get("return")
                if cls._is_response_model(return_annotation):
                    extra["response_model"] = return_annotation
            yield cls.get_api_action(
                endpoint,
                prefix=route_prefix,
                name=f"{endpoint.__name__} {cls.get_name()}",
                **extra,
            )

    @classmethod
    def get_action_detail_route(cls) -> str:
        """Detail-route prefix for ``@action(detail=True)`` endpoints."""
        return getattr(cls, "detail_route", "/{id}")

    @classmethod
    def get_api_action(
        cls,
        endpoint: Callable,
        prefix: str = "",
        path: str = "",
        **kwargs: Any,
    ) -> dict[str, Any]:
        kw = getattr(endpoint, "kwargs", {})
        generated_responses = kwargs.get("responses") or {}
        kwargs.update(kw)
        path = kwargs.get("path", path)
        kwargs["endpoint"] = endpoint
        kwargs["path"] = prefix + path
        kwargs.setdefault("name", endpoint.__name__)
        endpoint_name = kwargs["name"]
        kwargs.setdefault("methods", ["GET"])
        kwargs.setdefault("operation_id", f"{cls.get_slug_name()}_{endpoint_name}")
        kwargs["responses"] = _merge_responses(
            {e.get_status(): {"model": e.model} for e in cls.errors},
            _merge_responses(generated_responses, kw.get("responses") or {}),
        )
        status_code = kwargs.get("status_code")
        if status_code and not is_body_allowed_for_status_code(status_code):
            kwargs["response_model"] = None
        # ``detail`` (an ``@action`` marker applied to the path in
        # ``get_custom_api_actions``) and ``response_headers`` are not FastAPI
        # route arguments — consume them here so they never reach add_api_route.
        kwargs.pop("detail", None)
        response_headers = kwargs.pop("response_headers", None)
        if response_headers is not None:
            success = kwargs.get("status_code") or HTTP_200_OK
            responses = kwargs["responses"]
            # Copy before mutating: the per-status dict may be the very object
            # stored on the decorated method, shared across registrations.
            entry = {**responses.get(success, {})}
            entry["headers"] = {
                **entry.get("headers", {}),
                **response_headers.get_openapi_headers(),
            }
            responses[success] = entry
        return kwargs

finalize_response(response: Response) -> Response

Hook to post-process the built response before it is returned.

Returns it unchanged by default; mixins such as :class:~fastapi_views.views.mixins.ConditionalMixin override this to attach validators and downgrade to 304.

Source code in fastapi_views/views/api.py
def finalize_response(self, response: Response) -> Response:
    """Hook to post-process the built response before it is returned.

    Returns it unchanged by default; mixins such as
    :class:`~fastapi_views.views.mixins.ConditionalMixin` override this to
    attach validators and downgrade to ``304``.
    """
    return response

get_action_detail_route() -> str classmethod

Detail-route prefix for @action(detail=True) endpoints.

Source code in fastapi_views/views/api.py
@classmethod
def get_action_detail_route(cls) -> str:
    """Detail-route prefix for ``@action(detail=True)`` endpoints."""
    return getattr(cls, "detail_route", "/{id}")

APIView

Bases: View, ErrorHandlerMixin, Generic[T]

View with build-in json serialization via serializer and error handling

Source code in fastapi_views/views/api.py
class APIView(View, ErrorHandlerMixin, Generic[T]):
    """View with build-in json serialization via
    `serializer` and error handling
    """

    response_schema: T | None = None
    #: Extra route-level dependencies applied per action, e.g. auth scopes.
    action_dependencies: ClassVar[Mapping[Action, Sequence[params.Depends]]] = {}
    default_serializer_options: ClassVar[SerializerOptions] = {
        "by_alias": True,
    }
    default_errors: tuple[type[APIError], ...] = (BadRequest,)

    def __init__(self, request: Request, response: Response) -> None:
        self.validation_context = None
        self.serializer_options = self.default_serializer_options.copy()
        super().__init__(request, response)

    @classmethod
    def get_dependencies(cls, action: Action | None = None) -> list[params.Depends]:
        """Route-level dependencies for ``action``'s endpoint.

        Returns the :attr:`action_dependencies` entry for ``action``, e.g.
        auth scopes such as ``auth.requires("items:read")``. Override for
        fully dynamic per-action dependencies.
        """
        if action is None:
            return []
        return list(cls.action_dependencies.get(action, ()))

    @classmethod
    def get_response_headers(
        cls,
        action: Action | None = None,  # noqa: ARG003
    ) -> type[ResponseHeaders] | None:
        """Response headers to document in OpenAPI for the given ``action``.

        Override to declare headers (a :class:`~fastapi_views.models.ResponseHeaders`
        subclass) attached to the success response. Returns ``None`` by default.
        """
        return None

    @classmethod
    def get_conditional_responses(
        cls,
        *,
        action: Action | None = None,  # noqa: ARG003
        status_code: int | None = None,  # noqa: ARG003
        methods: Sequence[str] | None = None,  # noqa: ARG003
    ) -> dict[int | str, dict[str, Any]]:
        """Extra status-code responses contributed by mixins (e.g. ``304``).

        Returns an empty mapping by default; mixins such as
        :class:`~fastapi_views.views.mixins.ConditionalMixin` override this to
        document validator-driven responses.
        """
        return {}

    @classmethod
    def get_extra_responses(
        cls,
        *,
        action: Action | None = None,
        status_code: int | None = None,
        methods: Sequence[str] | None = None,
    ) -> dict[int | str, dict[str, Any]]:
        """Build the OpenAPI ``responses`` contributed by the view itself.

        Documents :meth:`get_response_headers` on the success status code and
        merges in any :meth:`get_conditional_responses`, combining the header
        maps when both target the same status code.
        """
        responses: dict[int | str, dict[str, Any]] = {}
        response_headers = cls.get_response_headers(action)
        if response_headers is not None and status_code is not None:
            responses[status_code] = {"headers": response_headers.get_openapi_headers()}
        conditional = cls.get_conditional_responses(
            action=action, status_code=status_code, methods=methods
        )
        for status, response in conditional.items():
            target = responses.setdefault(status, {})
            for key, value in response.items():
                if key == "headers" and "headers" in target:
                    target["headers"] = {**target["headers"], **value}
                else:
                    target[key] = value
        return responses

    @classmethod
    def get_api_action(
        cls,
        endpoint: Callable,
        prefix: str = "",
        path: str = "",
        action: Action | None = None,
        extra_errors: tuple[type[APIError], ...] = (),
        **kwargs: Any,
    ) -> dict[str, Any]:
        if action:
            kwargs.setdefault("name", f"{action.title()} {cls.get_name()}")
            kwargs.setdefault("operation_id", f"{action}_{cls.get_slug_name()}")

        kwargs.setdefault("response_model", cls.get_response_schema(action))

        dependencies = [
            *cls.get_dependencies(action),
            *(kwargs.get("dependencies") or ()),
        ]
        if dependencies:
            kwargs["dependencies"] = dependencies

        # A custom route's ``status_code`` / ``methods`` live on the decorated
        # method and only reach ``kwargs`` further down the MRO, so read them
        # from there too, or nothing gets documented for ``@action`` routes.
        route_options = getattr(endpoint, "kwargs", {})
        status_code = route_options.get("status_code") or kwargs.get("status_code")
        methods = route_options.get("methods") or kwargs.get("methods")
        extra_responses = cls.get_extra_responses(
            action=action,
            status_code=status_code or HTTP_200_OK,
            methods=methods or ["GET"],
        )
        kwargs["responses"] = _merge_responses(
            _merge_responses(
                errors(*extra_errors, *cls.default_errors),
                extra_responses,
            ),
            kwargs.get("responses") or {},
        )
        return super().get_api_action(endpoint, prefix=prefix, path=path, **kwargs)

    @classmethod
    def get_status_code(cls, endpoint: str, default: int = HTTP_200_OK) -> int:
        method = getattr(cls, endpoint, None)
        return getattr(method, "kwargs", {}).get("status_code", default)

    @classmethod
    def get_response_schema(cls, action: Action | None = None) -> T | None:  # noqa: ARG003
        return cls.response_schema

    def get_json_content(self, content: Any, serializer: TypeAdapter[Any]) -> bytes:
        if self.validate_response:
            content = serializer.validate_python(
                content,
                from_attributes=self.from_attributes,
                context=self.validation_context,
            )
            return serializer.dump_json(content, **self.serializer_options)
        return serializer.dump_json(content, warnings=False, **self.serializer_options)

get_dependencies(action: Action | None = None) -> list[params.Depends] classmethod

Route-level dependencies for action's endpoint.

Returns the :attr:action_dependencies entry for action, e.g. auth scopes such as auth.requires("items:read"). Override for fully dynamic per-action dependencies.

Source code in fastapi_views/views/api.py
@classmethod
def get_dependencies(cls, action: Action | None = None) -> list[params.Depends]:
    """Route-level dependencies for ``action``'s endpoint.

    Returns the :attr:`action_dependencies` entry for ``action``, e.g.
    auth scopes such as ``auth.requires("items:read")``. Override for
    fully dynamic per-action dependencies.
    """
    if action is None:
        return []
    return list(cls.action_dependencies.get(action, ()))

get_response_headers(action: Action | None = None) -> type[ResponseHeaders] | None classmethod

Response headers to document in OpenAPI for the given action.

Override to declare headers (a :class:~fastapi_views.models.ResponseHeaders subclass) attached to the success response. Returns None by default.

Source code in fastapi_views/views/api.py
@classmethod
def get_response_headers(
    cls,
    action: Action | None = None,  # noqa: ARG003
) -> type[ResponseHeaders] | None:
    """Response headers to document in OpenAPI for the given ``action``.

    Override to declare headers (a :class:`~fastapi_views.models.ResponseHeaders`
    subclass) attached to the success response. Returns ``None`` by default.
    """
    return None

get_conditional_responses(*, action: Action | None = None, status_code: int | None = None, methods: Sequence[str] | None = None) -> dict[int | str, dict[str, Any]] classmethod

Extra status-code responses contributed by mixins (e.g. 304).

Returns an empty mapping by default; mixins such as :class:~fastapi_views.views.mixins.ConditionalMixin override this to document validator-driven responses.

Source code in fastapi_views/views/api.py
@classmethod
def get_conditional_responses(
    cls,
    *,
    action: Action | None = None,  # noqa: ARG003
    status_code: int | None = None,  # noqa: ARG003
    methods: Sequence[str] | None = None,  # noqa: ARG003
) -> dict[int | str, dict[str, Any]]:
    """Extra status-code responses contributed by mixins (e.g. ``304``).

    Returns an empty mapping by default; mixins such as
    :class:`~fastapi_views.views.mixins.ConditionalMixin` override this to
    document validator-driven responses.
    """
    return {}

get_extra_responses(*, action: Action | None = None, status_code: int | None = None, methods: Sequence[str] | None = None) -> dict[int | str, dict[str, Any]] classmethod

Build the OpenAPI responses contributed by the view itself.

Documents :meth:get_response_headers on the success status code and merges in any :meth:get_conditional_responses, combining the header maps when both target the same status code.

Source code in fastapi_views/views/api.py
@classmethod
def get_extra_responses(
    cls,
    *,
    action: Action | None = None,
    status_code: int | None = None,
    methods: Sequence[str] | None = None,
) -> dict[int | str, dict[str, Any]]:
    """Build the OpenAPI ``responses`` contributed by the view itself.

    Documents :meth:`get_response_headers` on the success status code and
    merges in any :meth:`get_conditional_responses`, combining the header
    maps when both target the same status code.
    """
    responses: dict[int | str, dict[str, Any]] = {}
    response_headers = cls.get_response_headers(action)
    if response_headers is not None and status_code is not None:
        responses[status_code] = {"headers": response_headers.get_openapi_headers()}
    conditional = cls.get_conditional_responses(
        action=action, status_code=status_code, methods=methods
    )
    for status, response in conditional.items():
        target = responses.setdefault(status, {})
        for key, value in response.items():
            if key == "headers" and "headers" in target:
                target["headers"] = {**target["headers"], **value}
            else:
                target[key] = value
    return responses

AsyncListAPIView

Bases: BaseListAPIView, ABC, Generic[P]

Async list api view

Source code in fastapi_views/views/api.py
class AsyncListAPIView(BaseListAPIView, ABC, Generic[P]):
    """Async list api view"""

    @classmethod
    def get_list_endpoint(cls, status_code: int) -> Endpoint:
        schema = cls.get_response_schema(action="list")

        async def endpoint(
            self: AsyncListAPIView,
            *args: P.args,
            **kwargs: P.kwargs,
        ) -> Response:
            objects = await self.list(*args, **kwargs)
            return self.get_response(objects, status_code=status_code, schema=schema)

        cls._patch_endpoint_signature(endpoint, cls.list)
        return endpoint

    @abstractmethod
    async def list(self, *args: P.args, **kwargs: P.kwargs) -> Any:
        raise NotImplementedError

ListAPIView

Bases: BaseListAPIView, ABC, Generic[P]

Sync list api view

Source code in fastapi_views/views/api.py
class ListAPIView(BaseListAPIView, ABC, Generic[P]):
    """Sync list api view"""

    @classmethod
    def get_list_endpoint(cls, status_code: int) -> Endpoint:
        schema = cls.get_response_schema(action="list")

        def endpoint(self: ListAPIView, *args: P.args, **kwargs: P.kwargs) -> Response:
            objects = self.list(*args, **kwargs)
            return self.get_response(objects, status_code=status_code, schema=schema)

        cls._patch_endpoint_signature(endpoint, cls.list)
        return endpoint

    @abstractmethod
    def list(self, *args: P.args, **kwargs: P.kwargs) -> Any:
        raise NotImplementedError

RetrieveAPIView

Bases: BaseRetrieveAPIView, Generic[P]

Sync retrieve api view

Source code in fastapi_views/views/api.py
class RetrieveAPIView(BaseRetrieveAPIView, Generic[P]):
    """Sync retrieve api view"""

    @classmethod
    def get_retrieve_endpoint(cls, status_code: int) -> Endpoint:
        schema = cls.get_response_schema(action="retrieve")

        def endpoint(
            self: RetrieveAPIView,
            *args: P.args,
            **kwargs: P.kwargs,
        ) -> Response:
            obj = self.retrieve(*args, **kwargs)
            if obj is None and self.raise_on_none:
                self.raise_not_found_error()
            return self.get_response(obj, status_code=status_code, schema=schema)

        cls._patch_endpoint_signature(endpoint, cls.retrieve)
        return endpoint

    @abstractmethod
    def retrieve(self, *args: P.args, **kwargs: P.kwargs) -> Any:
        raise NotImplementedError

AsyncRetrieveAPIView

Bases: BaseRetrieveAPIView, Generic[P]

Async retrieve api view

Source code in fastapi_views/views/api.py
class AsyncRetrieveAPIView(BaseRetrieveAPIView, Generic[P]):
    """Async retrieve api view"""

    @classmethod
    def get_retrieve_endpoint(cls, status_code: int) -> Endpoint:
        schema = cls.get_response_schema(action="retrieve")

        async def endpoint(
            self: AsyncRetrieveAPIView,
            *args: P.args,
            **kwargs: P.kwargs,
        ) -> Response:
            obj = await self.retrieve(*args, **kwargs)
            if obj is None and self.raise_on_none:
                self.raise_not_found_error()
            return self.get_response(obj, status_code=status_code, schema=schema)

        cls._patch_endpoint_signature(endpoint, cls.retrieve)
        return endpoint

    @abstractmethod
    async def retrieve(self, *args: P.args, **kwargs: P.kwargs) -> Any:
        raise NotImplementedError

CreateAPIView

Bases: BaseCreateAPIView, Generic[P]

Sync create api view

Source code in fastapi_views/views/api.py
class CreateAPIView(BaseCreateAPIView, Generic[P]):
    """Sync create api view"""

    @classmethod
    def get_create_endpoint(cls, status_code: int) -> Endpoint:
        schema = cls.get_response_schema(action="create")

        def endpoint(
            self: CreateAPIView,
            *args: P.args,
            **kwargs: P.kwargs,
        ) -> Response:
            obj = self.create(*args, **kwargs)
            location = self.get_location(obj)
            if not self.return_on_create:
                obj = None
            return self.get_response(
                obj,
                status_code=status_code,
                schema=schema,
                headers={"location": location} if location else None,
            )

        cls._patch_endpoint_signature(endpoint, cls.create)
        return endpoint

    @abstractmethod
    def create(self, *args: P.args, **kwargs: P.kwargs) -> Any:
        raise NotImplementedError

AsyncCreateAPIView

Bases: BaseCreateAPIView, Generic[P]

Async create api view

Source code in fastapi_views/views/api.py
class AsyncCreateAPIView(BaseCreateAPIView, Generic[P]):
    """Async create api view"""

    @classmethod
    def get_create_endpoint(cls, status_code: int) -> Endpoint:
        schema = cls.get_response_schema(action="create")

        async def endpoint(
            self: AsyncCreateAPIView,
            *args: P.args,
            **kwargs: P.kwargs,
        ) -> Response:
            obj = await self.create(*args, **kwargs)
            location = self.get_location(obj)
            if not self.return_on_create:
                obj = None
            return self.get_response(
                obj,
                status_code=status_code,
                schema=schema,
                headers={"location": location} if location else None,
            )

        cls._patch_endpoint_signature(endpoint, cls.create)
        return endpoint

    @abstractmethod
    async def create(self, *args: P.args, **kwargs: P.kwargs) -> Any:
        raise NotImplementedError

UpdateAPIView

Bases: BaseUpdateAPIView, Generic[P]

Sync update api view

Source code in fastapi_views/views/api.py
class UpdateAPIView(BaseUpdateAPIView, Generic[P]):
    """Sync update api view"""

    @classmethod
    def get_update_endpoint(cls, status_code: int) -> Endpoint:
        schema = cls.get_response_schema(action="update")

        def endpoint(
            self: UpdateAPIView,
            *args: P.args,
            **kwargs: P.kwargs,
        ) -> Response:
            obj = self.update(*args, **kwargs)
            if obj is None and self.raise_on_none:
                self.raise_not_found_error()
            if not self.return_on_update:
                obj = None
            return self.get_response(obj, status_code=status_code, schema=schema)

        cls._patch_endpoint_signature(endpoint, cls.update)
        return endpoint

    @abstractmethod
    def update(self, *args: P.args, **kwargs: P.kwargs) -> Any:
        raise NotImplementedError

AsyncUpdateAPIView

Bases: BaseUpdateAPIView, Generic[P]

Async update api view

Source code in fastapi_views/views/api.py
class AsyncUpdateAPIView(BaseUpdateAPIView, Generic[P]):
    """Async update api view"""

    @classmethod
    def get_update_endpoint(cls, status_code: int) -> Endpoint:
        schema = cls.get_response_schema(action="update")

        async def endpoint(
            self: AsyncUpdateAPIView,
            *args: P.args,
            **kwargs: P.kwargs,
        ) -> Response:
            obj = await self.update(*args, **kwargs)
            if obj is None and self.raise_on_none:
                self.raise_not_found_error()
            if not self.return_on_update:
                obj = None
            return self.get_response(obj, status_code=status_code, schema=schema)

        cls._patch_endpoint_signature(endpoint, cls.update)
        return endpoint

    @abstractmethod
    async def update(self, *args: P.args, **kwargs: P.kwargs) -> Any:
        raise NotImplementedError

PartialUpdateAPIView

Bases: BasePartialUpdateAPIView, Generic[P]

Sync partial update api view

Source code in fastapi_views/views/api.py
class PartialUpdateAPIView(BasePartialUpdateAPIView, Generic[P]):
    """Sync partial update api view"""

    @classmethod
    def get_partial_update_endpoint(cls, status_code: int) -> Endpoint:
        schema = cls.get_response_schema(action="partial_update")

        def endpoint(
            self: PartialUpdateAPIView,
            *args: P.args,
            **kwargs: P.kwargs,
        ) -> Response:
            obj = self.partial_update(*args, **kwargs)
            if obj is None and self.raise_on_none:
                self.raise_not_found_error()
            if not self.return_on_update:
                obj = None
            return self.get_response(obj, status_code=status_code, schema=schema)

        cls._patch_endpoint_signature(endpoint, cls.partial_update)
        return endpoint

    @abstractmethod
    def partial_update(self, *args: P.args, **kwargs: P.kwargs) -> Any:
        raise NotImplementedError

AsyncPartialUpdateAPIView

Bases: BasePartialUpdateAPIView, Generic[P]

Async partial update api view

Source code in fastapi_views/views/api.py
class AsyncPartialUpdateAPIView(BasePartialUpdateAPIView, Generic[P]):
    """Async partial update api view"""

    @classmethod
    def get_partial_update_endpoint(cls, status_code: int) -> Endpoint:
        schema = cls.get_response_schema(action="partial_update")

        async def endpoint(
            self: AsyncPartialUpdateAPIView,
            *args: P.args,
            **kwargs: P.kwargs,
        ) -> Response:
            obj = await self.partial_update(*args, **kwargs)
            if obj is None and self.raise_on_none:
                self.raise_not_found_error()
            if not self.return_on_update:
                obj = None
            return self.get_response(obj, status_code=status_code, schema=schema)

        cls._patch_endpoint_signature(endpoint, cls.partial_update)
        return endpoint

    @abstractmethod
    async def partial_update(self, *args: P.args, **kwargs: P.kwargs) -> Any:
        raise NotImplementedError

DestroyAPIView

Bases: BaseDestroyAPIView, Generic[P]

Sync destroy api view

Source code in fastapi_views/views/api.py
class DestroyAPIView(BaseDestroyAPIView, Generic[P]):
    """Sync destroy api view"""

    @classmethod
    def get_destroy_endpoint(cls, status_code: int) -> Endpoint:
        def endpoint(
            self: DestroyAPIView,
            *args: P.args,
            **kwargs: P.kwargs,
        ) -> Response:
            self.destroy(*args, **kwargs)
            return Response(status_code=status_code)

        cls._patch_endpoint_signature(endpoint, cls.destroy)
        return endpoint

    @abstractmethod
    def destroy(self, *args: P.args, **kwargs: P.kwargs) -> None:
        raise NotImplementedError

AsyncDestroyAPIView

Bases: BaseDestroyAPIView, Generic[P]

Async destroy api view

Source code in fastapi_views/views/api.py
class AsyncDestroyAPIView(BaseDestroyAPIView, Generic[P]):
    """Async destroy api view"""

    @classmethod
    def get_destroy_endpoint(cls, status_code: int) -> Endpoint:
        async def endpoint(
            self: AsyncDestroyAPIView,
            *args: P.args,
            **kwargs: P.kwargs,
        ) -> Response:
            await self.destroy(*args, **kwargs)
            return Response(status_code=status_code)

        cls._patch_endpoint_signature(endpoint, cls.destroy)
        return endpoint

    @abstractmethod
    async def destroy(self, *args: P.args, **kwargs: P.kwargs) -> None:
        raise NotImplementedError

Mixins

Behavioural mixins shared by the view classes above. ConditionalMixin, which adds ETag / Last-Modified validators and 304 handling, is documented with Caching.

fastapi_views.views.mixins.DependencyMixin

Source code in fastapi_views/views/mixins.py
class DependencyMixin:
    @classmethod
    def get_extra_annotations(cls, action: str) -> dict[str, Any]:  # noqa: ARG003
        """Parameter-annotation overrides this class contributes for ``action``.

        Override in a subclass to inject real type objects (schemas,
        dependencies) into the endpoint signature of the given action. No
        ``super()`` call is needed: contributions from every class in the
        MRO are collected and merged, with the most derived class winning
        on duplicate parameter names.
        """
        return {}

    @classmethod
    def _collect_extra_annotations(cls, action: str) -> dict[str, Any]:
        annotations: dict[str, Any] = {}
        for base in reversed(cls.__mro__):
            hook = vars(base).get("get_extra_annotations")
            if isinstance(hook, classmethod):
                annotations |= hook.__func__(cls, action)
        return annotations

    @classmethod
    def _patch_endpoint_signature(
        cls,
        endpoint: Any,
        method: Callable,
    ) -> None:
        """Copy ``method``'s signature onto ``endpoint`` for FastAPI.

        The first parameter becomes the view dependency, and the extra
        annotations collected for the action (the method's name) override
        same-named parameters with real type objects, so generic views
        inject their configured schemas into the generated endpoint without
        ever mutating the view method.
        """
        annotations = cls._collect_extra_annotations(method.__name__)
        old_signature = inspect.signature(method)
        old_parameters: list[inspect.Parameter] = list(
            old_signature.parameters.values(),
        )
        old_first_parameter = old_parameters[0]
        new_first_parameter = old_first_parameter.replace(default=Depends(cls))
        new_parameters = [new_first_parameter] + [
            parameter.replace(
                kind=inspect.Parameter.KEYWORD_ONLY,
                annotation=annotations.get(parameter.name, parameter.annotation),
            )
            for parameter in old_parameters[1:]
        ]
        new_signature = old_signature.replace(parameters=new_parameters)
        endpoint.__signature__ = new_signature
        endpoint.__doc__ = method.__doc__
        endpoint.__name__ = method.__name__
        endpoint.kwargs = getattr(method, "kwargs", {})

get_extra_annotations(action: str) -> dict[str, Any] classmethod

Parameter-annotation overrides this class contributes for action.

Override in a subclass to inject real type objects (schemas, dependencies) into the endpoint signature of the given action. No super() call is needed: contributions from every class in the MRO are collected and merged, with the most derived class winning on duplicate parameter names.

Source code in fastapi_views/views/mixins.py
@classmethod
def get_extra_annotations(cls, action: str) -> dict[str, Any]:  # noqa: ARG003
    """Parameter-annotation overrides this class contributes for ``action``.

    Override in a subclass to inject real type objects (schemas,
    dependencies) into the endpoint signature of the given action. No
    ``super()`` call is needed: contributions from every class in the
    MRO are collected and merged, with the most derived class winning
    on duplicate parameter names.
    """
    return {}

fastapi_views.views.mixins.DetailViewMixin

Source code in fastapi_views/views/mixins.py
class DetailViewMixin:
    detail_route: str = "/{id}"
    raise_on_none: bool = True
    get_name: Callable[..., str]
    error_message = "{} does not exist"

    @classmethod
    def get_detail_route(cls, action: Action) -> str:  # noqa: ARG003
        return cls.detail_route

    def raise_not_found_error(self) -> NoReturn:
        msg = self.error_message.format(self.get_name())
        raise NotFound(msg)

fastapi_views.views.mixins.ErrorHandlerMixin

Source code in fastapi_views/views/mixins.py
class ErrorHandlerMixin:
    raises: ClassVar[dict[type[Exception], str | dict[str, Any]]] = {}

    def get_error_message(self, key: type[Exception]) -> str | dict[str, Any]:
        return self.raises.get(key, {})

    def handle_error(self, exc: Exception, **kwargs: Any) -> NoReturn:
        kw = self.get_error_message(type(exc))
        if isinstance(kw, str):
            kwargs["detail"] = kw
        elif isinstance(kw, Mapping):
            kwargs.update(kw)
        kwargs.setdefault("title", type(exc).__name__)
        kwargs.setdefault("detail", str(exc))
        kwargs.setdefault("status", HTTP_400_BAD_REQUEST)
        raise APIError(**kwargs)

    def get_exception_class(self) -> tuple[type[Exception], ...] | type[Exception]:
        return tuple(self.raises.keys()) or _Sentinel