0.0.1

Author: ProFastCode

app/__init__.py

@@ -1,6 +1,4 @@
-"""
-Init App Module
-"""
+import os
 
 from fastapi import FastAPI
 from fastapi.responses import JSONResponse
@@ -9,14 +7,16 @@
 from app.core import exps
 from app.core.settings import settings
 
+os.environ["TZ"] = "UTC"
+
 app = FastAPI(
-    title=settings.APP_TITLE,
-    root_path=settings.APP_PATH,
-    version=settings.APP_VERSION,
+    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",
     },
 )
 
@@ -27,5 +27,5 @@
 async def exception_handler(request, exc: exps.CustomException):
     return JSONResponse(
         status_code=exc.status_code,
-        content={'detail': exc.message},
+        content={"detail": exc.message},
     )

app/api/deps.py

@@ -5,10 +5,13 @@
 from typing import Annotated, AsyncGenerator
 
 from fastapi import Depends
-from fastapi.security import APIKeyHeader
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 
 from app.logic import Logic as _Logic
-from app.models.users import User as _User
+from app.models.user import User as _User
+
+
+security = HTTPBearer()
 
 
 async def get_logic() -> AsyncGenerator[_Logic, None]:
@@ -20,10 +23,10 @@ async def get_logic() -> AsyncGenerator[_Logic, None]:
 
 
 async def get_user(
-    token: Annotated[str, Depends(APIKeyHeader(name='access-token'))],
+    creds: Annotated[HTTPAuthorizationCredentials, Depends(security)],
     logic: Logic,
 ) -> _User | None:
-    return await logic.users.retrieve_by_token(token)
+    return await logic.user.retrieve_by_token(creds.credentials)
 
 
 User = Annotated[_User, Depends(get_user)]

app/api/v1/__init__.py

@@ -6,12 +6,12 @@
 
 from fastapi import APIRouter
 
-from . import auth, users
+from . import auth, user
 
-FOLDER_NAME = f'{Path(__file__).parent.name}'
+FOLDER_NAME = f"{Path(__file__).parent.name}"
 
-router = APIRouter(prefix=f'/{FOLDER_NAME}', tags=[FOLDER_NAME])
+router = APIRouter(prefix=f"/{FOLDER_NAME}", tags=[FOLDER_NAME])
 router.include_router(auth.router)
-router.include_router(users.router)
+router.include_router(user.router)
 
-__all__ = ['router']
+__all__ = ["router"]

app/api/v1/auth.py

@@ -0,0 +1,15 @@
+from fastapi import APIRouter
+
+from app.api import deps
+from app.models import auth as auth_models
+from app.models import user as user_models
+
+router = APIRouter(prefix="/auth")
+
+
+@router.post("/token", response_model=auth_models.AccessToken)
+async def token(data: user_models.UserCreate, logic: deps.Logic):
+    """
+    Retrieve new access token
+    """
+    return await logic.auth.generate_token(data)

app/api/v1/auth/__init__.py

@@ -0,0 +1,22 @@
+from fastapi import APIRouter
+
+from app.api import deps
+from app.models import user as user_models
+
+router = APIRouter(prefix="/users")
+
+
+@router.post("", response_model=user_models.UserRetrieve)
+async def create(data: user_models.UserCreate, logic: deps.Logic):
+    """
+    Create user
+    """
+    return await logic.user.create(data)
+
+
+@router.get("", response_model=user_models.UserRetrieve)
+async def retrieve(user: deps.User):
+    """
+    Retrieve user
+    """
+    return user

app/api/v1/auth/token.py

@@ -1,3 +0,0 @@
-"""
-Core App Module
-"""

app/api/v1/user.py

@@ -1,53 +1,56 @@
-"""
-Database
-"""
-
 from typing import Self
 
-from sqlalchemy.ext.asyncio import (AsyncEngine, async_sessionmaker,
-                                    create_async_engine)
+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:
+    user: repos.UserRepo
+
+    _instance = None
+
+    def __new__(cls, *args, **kwargs):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+        return cls._instance
+
     def __init__(
         self,
         engine: AsyncEngine | None = None,
-        session: AsyncSession | None = None,
+        session_maker: async_sessionmaker | None = None,
     ) -> None:
-        self.__engine = engine
-        self.__session = session
+        if not hasattr(self, "initialized"):
+            self.__engine = engine
+            self.__session_maker = session_maker
+            self.__initialized = True
 
     async def __set_async_engine(self) -> None:
         if self.__engine is None:
-            self.__engine = create_async_engine(
-                settings.db_dsn, echo=False, future=True
-            )
+            self.__engine = create_async_engine(settings.postgres_dsn)
 
-    async def __set_async_session(self) -> None:
-        if self.__session is None:
-            self.__session = async_sessionmaker(
+    async def __set_session_maker(self) -> None:
+        if self.__session_maker is None:
+            self.__session_maker = 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)
+        if self.__session_maker is not None:
+            self.user = repos.UserRepo(self.__session_maker)
 
     async def __aenter__(self) -> Self:
         await self.__set_async_engine()
-        await self.__set_async_session()
+        await self.__set_session_maker()
         await self.__set_repositories()
         return self
 
     async def __aexit__(self, exc_type, exc_value, traceback) -> None:
-        if self.__session is not None:
-            await self.__session.commit()
-            await self.__session.close()
+        if self.__engine is not None:
+            await self.__engine.dispose()

app/api/v1/users/__init__.py

@@ -1,8 +1,3 @@
-"""
-Exceptions
-"""
-
-
 class CustomException(Exception):
     def __init__(self, message: str, status_code: int = 500):
         super().__init__(message)
@@ -13,25 +8,25 @@ def __init__(self, message: str, status_code: int = 500):
 # Users
 class UserExistsException(CustomException):
     def __init__(self):
-        super().__init__('User is already taken.', status_code=409)
+        super().__init__("User is already taken.", status_code=409)
 
 
 class UserNotFoundException(CustomException):
     def __init__(self):
-        super().__init__('User not found.', status_code=404)
+        super().__init__("User not found.", status_code=404)
 
 
 class UserIsCorrectException(CustomException):
     def __init__(self):
-        super().__init__('User is correct.', status_code=401)
+        super().__init__("User is correct.", status_code=401)
 
 
 # Tokens
 class TokenInvalidException(CustomException):
     def __init__(self):
-        super().__init__('Invalid token.', status_code=401)
+        super().__init__("Invalid token.", status_code=401)
 
 
 class TokenExpiredException(CustomException):
     def __init__(self):
-        super().__init__('Token expired.', status_code=401)
+        super().__init__("Token expired.", status_code=401)

app/api/v1/users/create.py

@@ -1,39 +1,36 @@
-"""
-Settings
-"""
-
-from pydantic_settings import BaseSettings, SettingsConfigDict
 from sqlalchemy import URL
+from pydantic import Field
+from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
 class Settings(BaseSettings):
     model_config = SettingsConfigDict(
-        env_file='.env', env_file_encoding='utf-8', case_sensitive=True
+        env_file=".env", env_file_encoding="utf-8", case_sensitive=True
     )
 
-    # APP
-    APP_PATH: str = '/api'
-    APP_TITLE: str = 'FastAPI Template'
-    APP_VERSION: str = 'beta'
-    APP_SECRET_KEY: str = 'abc'
+    # App-related environment variables
+    app_path: str = Field(alias="APP_PATH")
+    app_title: str = Field(alias="APP_TITLE")
+    app_secret: str = Field(alias="APP_SECRET")
+    app_version: str = Field(alias="APP_VERSION")
 
-    # DATABASE
-    DB: str = 'postgres'
-    DB_HOST: str = 'localhost'
-    DB_PORT: int = 5432
-    DB_USER: str = ''
-    DB_PASSWORD: str = ''
-    DB_DRIVERNAME: str = 'postgresql+asyncpg'
+    # Postgres-related environment variables
+    postgres_db: str = Field(alias="POSTGRES_DB")
+    postgres_host: str = Field(alias="POSTGRES_HOST")
+    postgres_port: int = Field(alias="POSTGRES_PORT")
+    postgres_user: str = Field(alias="POSTGRES_USER")
+    postgres_password: str = Field(alias="POSTGRES_PASSWORD")
+    postgres_drivername: str = Field(alias="POSTGRES_DRIVERNAME")
 
     @property
-    def db_dsn(self) -> str:
+    def postgres_dsn(self) -> str:
         return URL.create(
-            drivername=self.DB_DRIVERNAME,
-            username=self.DB_USER,
-            password=self.DB_PASSWORD,
-            host=self.DB_HOST,
-            port=self.DB_PORT,
-            database=self.DB,
+            host=self.postgres_host,
+            port=self.postgres_port,
+            database=self.postgres_db,
+            username=self.postgres_user,
+            password=self.postgres_password,
+            drivername=self.postgres_drivername,
         ).render_as_string(hide_password=False)