Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 21 additions & 13 deletions src/fastrest/generics.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,12 @@

from __future__ import annotations

from typing import Any
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
from fastrest.compat.orm.base import ORMAdapter
from fastrest.pagination import BasePagination
from fastrest.filters import BaseFilterBackend

from fastrest.views import APIView
from fastrest.response import Response
Expand All @@ -17,26 +22,29 @@


class GenericAPIView(APIView):
queryset = None # Model class
serializer_class = None
queryset: type | None = None # Model class
serializer_class: type | None = None
lookup_field: str = "pk"
lookup_url_kwarg: str | None = None
dependencies: list = []
pagination_class = None
filter_backends = None
lookup_field_type: type = int
dependencies: list[Any] = []
pagination_class: type[BasePagination] | None = None
filter_backends: list[type[BaseFilterBackend]] | None = None

_session = None
_paginator = None
_adapter = None
# Session type is ORM-dependent (e.g. AsyncSession for SQLAlchemy).
# Can't type it here without coupling to a specific backend.
_session: Any = None
_paginator: BasePagination | None = None
_adapter: ORMAdapter | None = None

@property
def adapter(self):
def adapter(self) -> ORMAdapter:
if self._adapter is None:
from fastrest.compat.orm import get_default_adapter
self._adapter = get_default_adapter()
return self._adapter

def get_session(self):
def get_session(self) -> Any:
return self._session

def set_session(self, session: Any) -> None:
Expand Down Expand Up @@ -72,7 +80,7 @@ def get_serializer(self, *args: Any, **kwargs: Any) -> Any:
def get_serializer_class(self) -> type:
return self.serializer_class

def get_serializer_context(self) -> dict:
def get_serializer_context(self) -> dict[str, Any]:
return {
"request": getattr(self, "request", None),
"view": self,
Expand Down Expand Up @@ -101,7 +109,7 @@ def paginate_queryset(self, queryset: list) -> list | None:
self._paginator = cls()
return self._paginator.paginate_queryset(queryset, self.request, view=self)

def get_paginated_response(self, data: list) -> dict:
def get_paginated_response(self, data: list) -> dict[str, Any]:
return self._paginator.get_paginated_response(data)


Expand Down
3 changes: 2 additions & 1 deletion src/fastrest/routers.py
Original file line number Diff line number Diff line change
Expand Up @@ -174,10 +174,11 @@ def _make_action_endpoint(viewset_cls, actions, detail, request_model, pk_type):
if detail and request_model:
async def endpoint(request: FastAPIRequest, pk: int = 0, body=None) -> Any:
return await viewset_cls._dispatch_view(actions, {}, request, pk=pk, _body=body)
endpoint.__annotations__ = {'request': FastAPIRequest, 'pk': int, 'body': request_model, 'return': Any}
endpoint.__annotations__ = {'request': FastAPIRequest, 'pk': pk_type, 'body': request_model, 'return': Any}
elif detail:
async def endpoint(request: FastAPIRequest, pk: int) -> Any:
return await viewset_cls._dispatch_view(actions, {}, request, pk=pk)
endpoint.__annotations__['pk'] = pk_type
elif request_model:
async def endpoint(request: FastAPIRequest, body=None) -> Any:
return await viewset_cls._dispatch_view(actions, {}, request, _body=body)
Expand Down
3 changes: 3 additions & 0 deletions src/fastrest/views.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,9 @@


class APIView:
request: Request
kwargs: dict[str, Any]

permission_classes: list = [AllowAny]
authentication_classes: list = []
throttle_classes: list = []
Expand Down
7 changes: 5 additions & 2 deletions src/fastrest/viewsets.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ class ViewSetMixin:
action: str | None = None
basename: str | None = None
lookup_field_type: type = int
action_map: dict[str, str] = {}

@classmethod
def as_view(cls, actions: dict[str, str] | None = None, **initkwargs: Any):
Expand Down Expand Up @@ -51,7 +52,7 @@ async def view(request: FastAPIRequest) -> Any:
return view

@classmethod
def get_action_endpoints(cls, actions: dict[str, str], basename: str, serializer_class=None):
def get_action_endpoints(cls, actions: dict[str, str], basename: str, serializer_class: type | None = None):
"""Produce separate endpoint functions per action with full metadata."""
from fastrest.openapi import serializer_to_response_model, serializer_to_request_model

Expand Down Expand Up @@ -193,6 +194,7 @@ async def endpoint(request: FastAPIRequest) -> Any:
def _make_detail_endpoint(cls, actions: dict, pk_type: type):
async def endpoint(request: FastAPIRequest, pk: int) -> Any:
return await cls._dispatch_view(actions, {}, request, pk=pk)
endpoint.__annotations__['pk'] = pk_type
return endpoint

@classmethod
Expand All @@ -212,10 +214,11 @@ def _make_body_detail_endpoint(cls, actions: dict, request_model, pk_type: type)
if request_model:
async def endpoint(request: FastAPIRequest, pk: int = 0, body=None) -> Any:
return await cls._dispatch_view(actions, {}, request, pk=pk, _body=body)
endpoint.__annotations__ = {'request': FastAPIRequest, 'pk': int, 'body': request_model, 'return': Any}
endpoint.__annotations__ = {'request': FastAPIRequest, 'pk': pk_type, 'body': request_model, 'return': Any}
else:
async def endpoint(request: FastAPIRequest, pk: int) -> Any:
return await cls._dispatch_view(actions, {}, request, pk=pk)
endpoint.__annotations__['pk'] = pk_type
return endpoint

@classmethod
Expand Down