refactor: restore old functionality for gradual refactor

Author: stefanotroncaro

alembic/env.py

@@ -1,17 +1,14 @@
 from __future__ import with_statement
 
-from logging.config import fileConfig
-
-from sqlalchemy import engine_from_config
-from sqlalchemy import pool
+from alembic import context
+from sqlalchemy import engine_from_config, pool
 from sqlalchemy.sql.schema import SchemaItem
+from logging.config import fileConfig
 
-from alembic import context
-from app.common.infrastructure.base_sqlalchemy_model import BaseSQLAlchemyModel
 from app.core.config import get_settings
-from app.users.infrastructure.sqlalchemy_user_model import *  # noqa
-
-# from app.two_factor_authentication.models import *  # noqa
+from app.common.models.base_class import Base
+from app.users.models import *  # noqa
+from app.two_factor_authentication.models import *  # noqa
 
 
 # this is the Alembic Config object, which provides
@@ -26,7 +23,7 @@
 # add your model's MetaData object here
 # for 'autogenerate' support
 
-target_metadata = BaseSQLAlchemyModel.metadata
+target_metadata = Base.metadata
 # target_metadata = None
 
 

app/auth/services/auth_service.py

@@ -1,29 +1,24 @@
 from pydantic import EmailStr
 from sqlalchemy.orm import Session
 
-from app.auth.exceptions.invalid_credentials_exception import (
-    InvalidCredentialsException,
-)
 from app.auth.schemas.auth_schema import UserLogin
-from app.auth.utils.security import verify_password
 from app.common.exceptions.model_not_found_exception import (
     ModelNotFoundException,
 )
-from app.users.domain.user_dtos import UserAuth
-from app.users.domain.user_dtos import UserUpdate
-from app.users.infrastructure.sqlalchemy_user_repository import (
-    SQLAlchemyUserRepository,
+from app.auth.utils.security import verify_password
+from app.auth.exceptions.invalid_credentials_exception import (
+    InvalidCredentialsException,
 )
-from app.users.infrastructure.sqlalchemy_user_repository import (
+from app.users.repositories.users_repository import (
+    UsersRepository,
     users_repository,
 )
+from app.users.schemas.user_schema import UserAuth, UserUpdate
 
 
 class AuthService:
     def __init__(
-        self,
-        session: Session,
-        repository: SQLAlchemyUserRepository = users_repository,
+        self, session: Session, repository: UsersRepository = users_repository
     ):
         self.session = session
         self.repository = repository

app/celery/tasks/emails.py

@@ -1,12 +1,13 @@
 from uuid import UUID
-
-from app.common.schemas.pagination_schema import ListFilter
-from app.core.config import get_settings
-from app.db.session import SessionLocal
 from app.emails.exceptions.email_client_exception import EmailClientException
+from app.common.schemas.pagination_schema import ListFilter
 from app.emails.services.emails_service import EmailService
+from app.db.session import SessionLocal
 from app.main import celery
-from app.users.domain.user_dtos import UserInDB
+
+
+from app.core.config import get_settings
+from app.users.schemas.user_schema import UserInDB
 from app.users.services.users_service import UsersService
 
 settings = get_settings()

app/common/domain/__init__.py

@@ -1,9 +1,4 @@
-__all__ = (
-    "Criteria", 
-    "Entity", 
-    "Repository",
-)
-
-from .entities import Entity
-from .criteria import Criteria
-from .repository import Repository
+from .entities import *
+from .criteria import *
+from .paginator_criteria import *
+from .repository import *

app/common/models/__init__.py

@@ -0,0 +1 @@
+from .base_class import Base  # noqa

app/common/models/base_class.py

@@ -0,0 +1,40 @@
+from uuid import UUID, uuid4
+from datetime import datetime
+
+from sqlalchemy import MetaData, func, UUID as SQLA_UUID, DateTime
+from sqlalchemy.orm import (
+    DeclarativeBase,
+    Mapped,
+    mapped_column,
+)
+
+
+class Base(DeclarativeBase):
+    __abstract__ = True
+
+    metadata = MetaData(
+        naming_convention={
+            "ix": "ix_%(column_0_label)s",
+            "uq": "uq_%(table_name)s_%(column_0_name)s",
+            "ck": "ck_%(table_name)s_%(constraint_name)s",
+            "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
+            "pk": "pk_%(table_name)s",
+        }
+    )
+
+    type_annotation_map = {
+        UUID: SQLA_UUID,
+        datetime: DateTime(timezone=True),
+    }
+
+    id: Mapped[UUID] = mapped_column(
+        default=uuid4,
+        primary_key=True,
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        server_default=func.now(),
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        server_default=func.now(),
+        server_onupdate=func.now(),
+    )

app/common/repositories/__init__.py

@@ -0,0 +1,83 @@
+from math import ceil
+from typing import Any, Generic, Optional, Type, TypeVar
+from uuid import UUID
+
+from fastapi.encoders import jsonable_encoder
+from pydantic import BaseModel
+from sqlalchemy import asc, desc
+from sqlalchemy.orm import Session
+from sqlalchemy.orm.query import Query
+
+from app.common.schemas.pagination_schema import ListFilter, ListResponse
+
+
+ModelType = TypeVar("ModelType", bound=Any)
+CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
+UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
+
+
+class BaseRepository(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
+    def __init__(self, model: Type[ModelType]):
+        """
+        CRUD object with default methods to Create, Read, Update, Delete (CRUD).
+
+        **Parameters**
+
+        * `model`: A SQLAlchemy model class
+        * `schema`: A Pydantic model (schema) class
+        """
+        self.model = model
+
+    def get(self, db: Session, model_id: UUID) -> Optional[ModelType]:
+        return db.query(self.model).filter(self.model.id == model_id).first()
+
+    def list(
+        self, db: Session, list_options: ListFilter, query: Query | None = None
+    ) -> ListResponse:
+        if not query:
+            query = db.query(self.model)
+
+        total = query.count()
+
+        if list_options.order_by:
+            column = list_options.order_by
+            direction = list_options.order
+            by = desc if direction == "desc" else asc
+
+            query = query.order_by(by(column))
+
+        query = query.offset(list_options.page_size * (list_options.page - 1))
+
+        query = query.limit(list_options.page_size)
+        return ListResponse(
+            data=query.all(),
+            page=list_options.page,
+            page_size=list_options.page_size,
+            total=total,
+            total_pages=ceil(total / list_options.page_size),
+        )
+
+    def create(self, db: Session, obj_in: CreateSchemaType) -> ModelType:
+        obj_in_data = jsonable_encoder(obj_in)
+        db_obj = self.model(**obj_in_data)  # type: ignore
+        db.add(db_obj)
+        db.flush()
+        return db_obj
+
+    def update(
+        self, db: Session, db_obj: ModelType, obj_in: UpdateSchemaType
+    ) -> ModelType:
+        obj_data = jsonable_encoder(db_obj)
+        update_data = obj_in.dict(exclude_unset=True)
+        for field in obj_data:
+            if field in update_data:
+                setattr(db_obj, field, update_data[field])
+        db.add(db_obj)
+        db.flush()
+        return db_obj
+
+    def delete(self, db: Session, model_id: UUID) -> ModelType | None:
+        obj = db.query(self.model).get(model_id)
+        db.delete(obj)
+        db.flush()
+        return obj

app/common/repositories/base_repository.py

@@ -0,0 +1 @@
+USER_EMAIL_MAX_LENGTH = 100

app/users/constants/user_constants.py

@@ -0,0 +1 @@
+from .user import User  # noqa