typing

Author: ProFastCode

app/core/db.py

@@ -1,3 +1,9 @@
+"""
+Database
+"""
+
+from typing import NoReturn, Self
+
 from sqlalchemy.ext.asyncio import (AsyncEngine, async_sessionmaker,
                                     create_async_engine)
 from sqlmodel.ext.asyncio.session import AsyncSession
@@ -9,7 +15,7 @@
 class Database:
     _instance = None
 
-    def __new__(cls, *args, **kwargs):
+    def __new__(cls, *args, **kwargs) -> Self:
         if cls._instance is None:
             cls._instance = super(Database, cls).__new__(cls)
         return cls._instance
@@ -24,13 +30,13 @@ def __init__(
             self.session = session
             self.initialized = True
 
-    async def __set_async_engine(self) -> None:
+    async def __set_async_engine(self) -> NoReturn:
         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:
+    async def __set_async_session(self) -> NoReturn:
         if self.session is None:
             self.session = async_sessionmaker(
                 autocommit=False,
@@ -40,16 +46,16 @@ async def __set_async_session(self) -> None:
                 expire_on_commit=False,
             )()
 
-    async def __set_repositories(self) -> None:
+    async def __set_repositories(self) -> NoReturn:
         if self.session is not None:
             self.user = repos.UserRepo(session=self.session)
 
-    async def __aenter__(self):
+    async def __aenter__(self) -> Self:
         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):
+    async def __aexit__(self, exc_type, exc_value, traceback) -> NoReturn:
         if self.session is not None:
             await self.session.close()

app/core/settings.py

@@ -26,15 +26,14 @@ class Settings(BaseSettings):
 
     @property
     def pg_dsn(self) -> PostgresDsn:
-        dsn = PostgresDsn.build(
+        return PostgresDsn.build(
             scheme='postgresql+asyncpg',
             username=self.POSTGRES_USER,
             password=self.POSTGRES_PASSWORD,
             host=self.POSTGRES_HOST,
             port=self.POSTGRES_PORT,
             path=self.POSTGRES_DB,
         )
-        return dsn
 
 
 settings = Settings()

app/logic/__init__.py

@@ -1,3 +1,5 @@
+from typing import Self
+
 from app.core.db import Database
 
 from .security import Security
@@ -11,7 +13,7 @@ def __init__(self, db: Database):
         self.users = Users(self)
 
     @classmethod
-    async def create(cls) -> 'Logic':
+    async def create(cls) -> Self:
         async with Database() as db:
             return cls(db)
 

app/logic/security/jwt.py

@@ -13,11 +13,11 @@ def decode_token(self, token: str) -> dict | None:
         try:
             payload = jwt.decode(token, self.secret_key, algorithms=['HS256'])
         except Exception:
-            raise exps.TOKEN_INVALID
+            raise exps.TokenInvalidException()
 
         exp = payload.get('exp')
         if exp and dt.datetime.now(dt.UTC).timestamp() > exp:
-            raise exps.TOKEN_EXPIRED
+            raise exps.TokenExpiredException()
         return payload.get('payload')
 
     def encode_token(self, payload: dict, minutes: int) -> str: