Логика не зависит от фреймворка, осталось решить зависимость от exps fastapi

Author: ProFastCode

app/api/anotations.py

@@ -0,0 +1,28 @@
+"""
+Dependencies
+"""
+
+from typing import Annotated
+
+from fastapi import Depends
+from fastapi.security import APIKeyHeader
+
+from app.logic import Logic as _Logic
+from app.models.user import User as _User
+
+
+async def get_logic() -> _Logic:
+    return await _Logic.create()
+
+
+Logic = Annotated[_Logic, Depends(get_logic)]
+
+
+async def get_user(
+    token: Annotated[str, Depends(APIKeyHeader(name="access-token"))],
+    logic: Logic,
+) -> _User | None:
+    return await logic.users.retrieve_by_token(token)
+
+
+User = Annotated[_User, Depends(get_user)]

app/api/deps.py

@@ -1,14 +1,14 @@
 from fastapi import APIRouter
 
-from app.api import anotations
+from app.api import deps
 from app.models.token import AccessToken
 from app.models.user import UserCreate
 
 router = APIRouter(prefix="/token")
 
 
 @router.post("/", response_model=AccessToken)
-async def token(data: UserCreate, logic: anotations.Logic):
+async def token(data: UserCreate, logic: deps.Logic):
     """
     Retrieve new access token
     """

app/api/v1/users/auth/token.py

@@ -1,13 +1,13 @@
 from fastapi import APIRouter
 
-from app.api import anotations
+from app.api import deps
 from app.models.user import UserCreate, UserRead
 
 router = APIRouter(prefix="/create")
 
 
 @router.post("/", response_model=UserRead)
-async def create(data: UserCreate, logic: anotations.Logic):
+async def create(data: UserCreate, logic: deps.Logic):
     """
     Create user
     """

app/api/v1/users/create.py

@@ -4,14 +4,14 @@
 
 from fastapi import APIRouter
 
-from app.api import anotations
+from app.api import deps
 from app.models.user import UserRead
 
 router = APIRouter()
 
 
 @router.get("/", response_model=UserRead)
-async def retrieve(user: anotations.CurrentUser):
+async def retrieve(user: deps.User):
     """
     Retrieve user
     """

app/api/v1/users/retrieve.py

@@ -10,32 +10,38 @@
 
 
 class Database:
-    session: AsyncSession
-
-    @classmethod
-    def __get_async_engine(cls) -> AsyncEngine:
-        return create_async_engine(
-            settings.pg_dsn.unicode_string(), echo=False, future=True
-        )
-
-    @classmethod
-    def __get_async_session(cls) -> async_sessionmaker[AsyncSession]:
-        return async_sessionmaker(
-            autocommit=False,
-            autoflush=False,
-            bind=cls.__get_async_engine(),
-            class_=AsyncSession,
-            expire_on_commit=False,
-        )
-
-    def __set_repositories(self):
-        self.user = repos.UserRepo(session=self.session)
+    def __init__(
+        self, engine: AsyncEngine | None = None, session: AsyncSession | None = None
+    ) -> None:
+        self.engine = engine
+        self.session = session
+
+    async def __set_async_engine(self) -> None:
+        if self.engine is None:
+            self.engine = create_async_engine(
+                settings.pg_dsn.unicode_string(), echo=False, future=True
+            )
+
+    async def __set_async_session(self) -> None:
+        if self.session is None:
+            self.session = async_sessionmaker(
+                autocommit=False,
+                autoflush=False,
+                bind=self.engine,
+                class_=AsyncSession,
+                expire_on_commit=False,
+            )()
+
+    async def __set_repositories(self) -> None:
+        if self.session is not None:
+            self.user = repos.UserRepo(session=self.session)
 
     async def __aenter__(self):
-        async with self.__get_async_session()() as session:
-            self.session = session
-        self.__set_repositories()
+        await self.__set_async_engine()
+        await self.__set_async_session()
+        await self.__set_repositories()
         return self
 
     async def __aexit__(self, exc_type, exc_value, traceback):
-        await self.session.close()
+        if self.session is not None:
+            await self.session.close()

app/core/db.py

@@ -1,12 +1,19 @@
 from app.core.db import Database
-from app.core.security import Security
 
 from .users import Users
+from .security import Security
 
 
 class Logic:
-    def __init__(self, db: Database, security: Security):
-        self.users = Users(db, security)
+    def __init__(self, db: Database):
+        self.db = db
+        self.security = Security()
+        self.users = Users(self)
 
+    @classmethod
+    async def create(cls):
+        async with Database() as db:
+            return cls(db)
 
-__all__ = ['Logic']
+
+__all__ = ["Logic"]