Department of Logic

Author: ProFastCode

app/api/anotations.py

@@ -1,14 +1,16 @@
 """
-Anotations
+Dependencies
 """
 
 from fastapi import Depends
 from typing_extensions import Annotated
 
 from app.core import deps
 
-Database = Annotated[deps.db.Database, Depends(deps.get_db)]
+Database = Annotated[deps.Database, Depends(deps.get_db)]
 
 Security = Annotated[deps.Security, Depends(deps.get_security)]
 
+Logic = Annotated[deps.Logic, Depends(deps.get_logic)]
+
 CurrentUser = Annotated[deps.User, Depends(deps.get_current_user)]

app/api/endpoints/users/auth.py

@@ -1,24 +1,15 @@
 from fastapi import APIRouter
 
 from app.api import anotations
-from app.core import exps
 from app.models.token import AccessToken
 from app.models.user import UserCreate
 
 router = APIRouter(prefix='/auth')
 
 
 @router.post('/token/', response_model=AccessToken)
-async def token(
-    data: UserCreate, db: anotations.Database, security: anotations.Security
-):
+async def token(data: UserCreate, logic: anotations.Logic):
     """
     Retrieve new access token
     """
-    if user := await db.user.retrieve_by_email(data.email):
-        if not security.pwd.checkpwd(data.password, user.password):
-            raise exps.USER_IS_CORRECT
-        access_token = security.jwt.encode_token({'id': user.id}, 1440)
-        return AccessToken(token=access_token)
-
-    raise exps.USER_NOT_FOUND
+    return await logic.users.generate_token(**data.model_dump())

app/api/endpoints/users/create.py

@@ -1,23 +1,14 @@
 from fastapi import APIRouter
 
 from app.api import anotations
-from app.core import exps
-from app.models.user import User, UserCreate, UserRead
+from app.models.user import UserCreate, UserRead
 
 router = APIRouter(prefix='/create')
 
 
 @router.post('/', response_model=UserRead)
-async def registration(
-    data: UserCreate, db: anotations.Database, security: anotations.Security
-):
+async def create(data: UserCreate, logic: anotations.Logic):
     """
     Create user
     """
-    if await db.user.retrieve_by_email(data.email):
-        raise exps.USER_EXISTS
-
-    password_hash = security.pwd.hashpwd(data.password)
-    model = User(email=data.email, password=password_hash)
-    user = await db.user.create(model)
-    return user
+    return await logic.users.create(**data.model_dump())

app/core/__init__.py

@@ -1,7 +1,3 @@
 """
 Core App Module
 """
-
-from .security import Security
-
-__all__ = ['Security']

app/core/deps.py

@@ -6,27 +6,30 @@
 from fastapi.security import APIKeyHeader
 from typing_extensions import Annotated, AsyncGenerator
 
+from app.core.db import Database, SessionLocal
+from app.core.logic import Logic
+from app.core.security import Security
 from app.models.user import User
 
-from . import Security, db, exps
 
-
-async def get_db() -> AsyncGenerator[db.Database]:
-    async with db.SessionLocal() as session:
-        yield db.Database(session)
+async def get_db() -> AsyncGenerator[Database]:
+    async with SessionLocal() as session:
+        yield Database(session)
 
 
 async def get_security() -> Security:
     return Security()
 
 
+async def get_logic(
+    db: Annotated[Database, Depends(get_db)],
+    security: Annotated[Security, Depends(get_security)],
+) -> Logic:
+    return Logic(db, security)
+
+
 async def get_current_user(
     token: Annotated[str, Depends(APIKeyHeader(name='access-token'))],
-    security: Annotated[Security, Depends(get_security)],
-    db: Annotated[db.Database, Depends(get_db)],
+    logic: Annotated[Logic, Depends(get_logic)],
 ) -> User | None:
-    if payload := security.jwt.decode_token(token):
-        if not (user := await db.user.retrieve_one(ident=payload.get('id'))):
-            raise exps.USER_NOT_FOUND
-        else:
-            return user
+    return await logic.users.retrieve_by_token(token)

app/core/logic.py

@@ -0,0 +1,11 @@
+from app.core.db import Database
+from app.core.security import Security
+from app.logic.users import Users
+
+
+class Logic:
+    def __init__(self, db: Database, security: Security):
+        self.users = Users(db, security)
+
+
+__all__ = ['Logic']

app/logic/users.py

@@ -0,0 +1,43 @@
+from app.core import exps
+from app.core.db import Database
+from app.core.security import Security
+from app.models.token import AccessToken
+from app.models.user import User
+
+
+class Users:
+    def __init__(self, db: Database, security: Security):
+        self.db = db
+        self.security = security
+
+    async def create(self, email: str, password: str) -> User | None:
+        if await self.db.user.retrieve_by_email(email):
+            raise exps.USER_EXISTS
+
+        password_hash = self.security.pwd.hashpwd(password)
+        model = User(email=email, password=password_hash)
+        user = await self.db.user.create(model)
+        return user
+
+    async def generate_token(
+        self, email: str, password: str
+    ) -> AccessToken | None:
+        if user := await self.db.user.retrieve_by_email(email):
+            if not self.security.pwd.checkpwd(password, user.password):
+                raise exps.USER_IS_CORRECT
+            access_token = self.security.jwt.encode_token(
+                {'id': user.id}, 1440
+            )
+            return AccessToken(token=access_token)
+        raise exps.USER_NOT_FOUND
+
+    async def retrieve_by_token(self, token: str) -> User | None:
+        if payload := self.security.jwt.decode_token(token):
+            if not (
+                user := await self.db.user.retrieve_one(
+                    ident=payload.get('id')
+                )
+            ):
+                raise exps.USER_NOT_FOUND
+            else:
+                return user