Database sington

Author: ProFastCode

app/__init__.py

@@ -3,19 +3,29 @@
 """
 
 from fastapi import FastAPI
+from fastapi.responses import JSONResponse
 
 from app import api
+from app.core import exps
 from app.core.settings import settings
 
 app = FastAPI(
     title=settings.APP_TITLE,
     root_path=settings.APP_PATH,
     version=settings.APP_VERSION,
     contact={
-        'name': 'Fast Code',
-        'url': 'https://fast-code.pro/',
-        'email': 'fast.code.auth@gmail.com',
+        "name": "Fast Code",
+        "url": "https://fast-code.pro/",
+        "email": "fast.code.auth@gmail.com",
     },
 )
 
 app.include_router(api.api_router)
+
+
+@app.exception_handler(exps.BaseException)
+async def exception_handler(request, exc: exps.BaseException):
+    return JSONResponse(
+        status_code=exc.status_code,
+        content={"detail": exc.message},
+    )

app/core/db.py

@@ -1,20 +1,24 @@
-"""
-Database
-"""
-
 from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine
 from sqlmodel.ext.asyncio.session import AsyncSession
-
 from app import repositories as repos
 from app.core.settings import settings
 
 
 class Database:
+    _instance = None
+
+    def __new__(cls, *args, **kwargs):
+        if cls._instance is None:
+            cls._instance = super(Database, cls).__new__(cls)
+        return cls._instance
+
     def __init__(
         self, engine: AsyncEngine | None = None, session: AsyncSession | None = None
     ) -> None:
-        self.engine = engine
-        self.session = session
+        if not hasattr(self, "initialized"):  # Проверка, инициализирован ли объект
+            self.engine = engine
+            self.session = session
+            self.initialized = True  # Установим флаг инициализации
 
     async def __set_async_engine(self) -> None:
         if self.engine is None:

app/core/exps.py

@@ -2,15 +2,36 @@
 Exceptions
 """
 
-from fastapi import HTTPException, status
+
+class BaseException(Exception):
+    def __init__(self, message: str, status_code: int = 500):
+        super().__init__(message)
+        self.message = message
+        self.status_code = status_code
+
 
 # Users
-USER_EXISTS = HTTPException(status.HTTP_409_CONFLICT, 'User is already taken.')
-USER_NOT_FOUND = HTTPException(status.HTTP_404_NOT_FOUND, 'User not found.')
-USER_IS_CORRECT = HTTPException(
-    status.HTTP_401_UNAUTHORIZED, 'User is correct.'
-)
+class UserExistsException(BaseException):
+    def __init__(self):
+        super().__init__("User is already taken.", status_code=409)
+
+
+class UserNotFoundException(BaseException):
+    def __init__(self):
+        super().__init__("User not found.", status_code=404)
+
+
+class UserIsCorrectException(BaseException):
+    def __init__(self):
+        super().__init__("User is correct.", status_code=401)
+
 
 # Tokens
-TOKEN_INVALID = HTTPException(status.HTTP_401_UNAUTHORIZED, 'Invalid token.')
-TOKEN_EXPIRED = HTTPException(status.HTTP_401_UNAUTHORIZED, 'Token expired.')
+class TokenInvalidException(BaseException):
+    def __init__(self):
+        super().__init__("Invalid token.", status_code=401)
+
+
+class TokenExpiredException(BaseException):
+    def __init__(self):
+        super().__init__("Token expired.", status_code=401)

app/logic/__init__.py

@@ -11,7 +11,7 @@ def __init__(self, db: Database):
         self.users = Users(self)
 
     @classmethod
-    async def create(cls):
+    async def create(cls) -> "Logic":
         async with Database() as db:
             return cls(db)
 

app/logic/users.py

@@ -15,7 +15,7 @@ def __init__(self, logic: "Logic"):
 
     async def create(self, email: str, password: str) -> User | None:
         if await self.logic.db.user.retrieve_by_email(email):
-            raise exps.USER_EXISTS
+            raise exps.UserExistsException()
 
         password_hash = self.logic.security.pwd.hashpwd(password)
         model = User(email=email, password=password_hash)
@@ -25,17 +25,16 @@ async def create(self, email: str, password: str) -> User | None:
     async def generate_token(self, email: str, password: str) -> AccessToken | None:
         if user := await self.logic.db.user.retrieve_by_email(email):
             if not self.logic.security.pwd.checkpwd(password, user.password):
-                raise exps.USER_IS_CORRECT
-            access_token = self.logic.security.jwt.encode_token(
-                {"id": user.id}, 1440)
+                raise exps.UserIsCorrectException()
+            access_token = self.logic.security.jwt.encode_token({"id": user.id}, 1440)
             return AccessToken(token=access_token)
-        raise exps.USER_NOT_FOUND
+        raise exps.UserNotFoundException()
 
     async def retrieve_by_token(self, token: str) -> User | None:
         if payload := self.logic.security.jwt.decode_token(token):
             if not (
                 user := await self.logic.db.user.retrieve_one(ident=payload.get("id"))
             ):
-                raise exps.USER_NOT_FOUND
+                raise exps.UserNotFoundException()
             else:
                 return user