Extract template service

The goal is to remove fastapi dependency
from service layer.

But we need to understand templates require some FastAPI
thins, so for templates (views) services to have fastapi
dependencies is fine.

Author: koldakov

futuramaapi/routers/services/__init__.py

@@ -1,9 +1,9 @@
 from ._base import (
     BaseService,
     BaseSessionService,
-    BaseTemplateService,
     BaseUserAuthenticatedService,
 )
+from ._base_template import BaseTemplateService
 
 __all__ = [
     "BaseService",

futuramaapi/routers/services/_base.py

@@ -1,25 +1,19 @@
 from abc import ABC, abstractmethod
 from collections.abc import Sequence
-from typing import Any, ClassVar, TypeVar
+from typing import Any, TypeVar
 
 import jwt
-from fastapi import HTTPException, Request, status
+from fastapi import HTTPException, status
 from fastapi_pagination import Page
 from jwt import ExpiredSignatureError, InvalidSignatureError, InvalidTokenError
-from pydantic import Field
-from sqlalchemy import Result, Select, select
+from sqlalchemy import Select, select
 from sqlalchemy.exc import NoResultFound
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy.orm import selectinload
-from starlette.templating import _TemplateResponse
 
-from futuramaapi.__version__ import __version__
 from futuramaapi.core import settings
 from futuramaapi.helpers.pydantic import BaseModel
-from futuramaapi.helpers.templates import templates
-from futuramaapi.repositories.models import AuthSessionModel, UserModel
+from futuramaapi.repositories.models import UserModel
 from futuramaapi.repositories.session import session_manager
-from futuramaapi.utils import config, metadata
 
 TResponse = TypeVar(
     "TResponse",
@@ -58,80 +52,6 @@ async def __call__(self, *args, **kwargs) -> TResponse:
             return await self.process(*args, **kwargs)
 
 
-class _ProjectContext(BaseModel):
-    version: str = Field(
-        default=__version__,
-    )
-    g_tag: str = Field(
-        default=settings.g_tag,
-        alias="G_TAG",
-    )
-    author: str | None = Field(
-        default=metadata.get("author", None),
-    )
-    description: str = Field(
-        default=metadata["summary"],
-    )
-    config: dict[str, Any] = config
-
-
-_project_context: _ProjectContext = _ProjectContext()
-
-
-class BaseTemplateService(BaseSessionService[_TemplateResponse], ABC):
-    template_name: ClassVar[str]
-
-    _cookie_auth_key: ClassVar[str] = "Authorization"
-
-    @abstractmethod
-    async def get_context(self, *args, **kwargs) -> dict[str, Any]: ...
-
-    @property
-    def request(self) -> Request:
-        if self.context is None:
-            raise AttributeError("Request is not defined.")
-
-        if "request" not in self.context:
-            raise AttributeError("Request is not defined.")
-
-        return self.context["request"]
-
-    @property
-    def __auth_session_statement(self) -> Select[tuple[AuthSessionModel]]:
-        statement: Select[tuple[AuthSessionModel]] = select(AuthSessionModel)
-        statement = statement.where(AuthSessionModel.key == self.request.cookies[self._cookie_auth_key])
-        statement = statement.options(selectinload(AuthSessionModel.user))
-        return statement
-
-    async def _get_current_user(self) -> UserModel | None:
-        if self._cookie_auth_key not in self.request.cookies:
-            return None
-
-        result: Result[tuple[AuthSessionModel]] = await self.session.execute(self.__auth_session_statement)
-        try:
-            auth_session: AuthSessionModel = result.scalars().one()
-        except NoResultFound:
-            return None
-
-        if not auth_session.valid:
-            return None
-
-        return auth_session.user
-
-    async def _get_context(self) -> dict[str, Any]:
-        context: dict[str, Any] = await self.get_context()
-        context["_project"] = _project_context
-        context["current_user"] = await self._get_current_user()
-        return context
-
-    async def process(self, *args, **kwargs) -> _TemplateResponse:
-        return templates.TemplateResponse(
-            self.request,
-            self.template_name,
-            context=await self._get_context(),
-        )
-
-
 class BaseUserAuthenticatedService[TResponse](BaseSessionService[TResponse], ABC):
     token: str
 

futuramaapi/routers/services/_base_template.py

@@ -0,0 +1,92 @@
+from abc import ABC, abstractmethod
+from typing import Any, ClassVar
+
+from fastapi import Request
+from pydantic import Field
+from sqlalchemy import Result, Select, select
+from sqlalchemy.exc import NoResultFound
+from sqlalchemy.orm import selectinload
+from starlette.templating import _TemplateResponse
+
+from futuramaapi.__version__ import __version__
+from futuramaapi.core import settings
+from futuramaapi.helpers.pydantic import BaseModel
+from futuramaapi.helpers.templates import templates
+from futuramaapi.repositories.models import AuthSessionModel, UserModel
+from futuramaapi.utils import config, metadata
+
+from ._base import BaseSessionService
+
+
+class _ProjectContext(BaseModel):
+    version: str = Field(
+        default=__version__,
+    )
+    g_tag: str = Field(
+        default=settings.g_tag,
+        alias="G_TAG",
+    )
+    author: str | None = Field(
+        default=metadata.get("author", None),
+    )
+    description: str = Field(
+        default=metadata["summary"],
+    )
+    config: dict[str, Any] = config
+
+
+_project_context: _ProjectContext = _ProjectContext()
+
+
+class BaseTemplateService(BaseSessionService[_TemplateResponse], ABC):
+    template_name: ClassVar[str]
+
+    _cookie_auth_key: ClassVar[str] = "Authorization"
+
+    @abstractmethod
+    async def get_context(self, *args, **kwargs) -> dict[str, Any]: ...
+
+    @property
+    def request(self) -> Request:
+        if self.context is None:
+            raise AttributeError("Request is not defined.")
+
+        if "request" not in self.context:
+            raise AttributeError("Request is not defined.")
+
+        return self.context["request"]
+
+    @property
+    def __auth_session_statement(self) -> Select[tuple[AuthSessionModel]]:
+        statement: Select[tuple[AuthSessionModel]] = select(AuthSessionModel)
+        statement = statement.where(AuthSessionModel.key == self.request.cookies[self._cookie_auth_key])
+        statement = statement.options(selectinload(AuthSessionModel.user))
+        return statement
+
+    async def _get_current_user(self) -> UserModel | None:
+        if self._cookie_auth_key not in self.request.cookies:
+            return None
+
+        result: Result[tuple[AuthSessionModel]] = await self.session.execute(self.__auth_session_statement)
+        try:
+            auth_session: AuthSessionModel = result.scalars().one()
+        except NoResultFound:
+            return None
+
+        if not auth_session.valid:
+            return None
+
+        return auth_session.user
+
+    async def _get_context(self) -> dict[str, Any]:
+        context: dict[str, Any] = await self.get_context()
+        context["_project"] = _project_context
+        context["current_user"] = await self._get_current_user()
+        return context
+
+    async def process(self, *args, **kwargs) -> _TemplateResponse:
+        return templates.TemplateResponse(
+            self.request,
+            self.template_name,
+            context=await self._get_context(),
+        )

futuramaapi/routers/services/passwords/get_signature_user_password_change_form.py

@@ -10,7 +10,7 @@
 from futuramaapi.helpers.templates import templates
 from futuramaapi.repositories.models import UserModel
 from futuramaapi.routers.services import BaseSessionService
-from futuramaapi.routers.services._base import _project_context
+from futuramaapi.routers.services._base_template import _project_context
 
 
 class TokenDecodeError(Exception):