✨ Add (module): Database CRUD operations and SA model representation

Author: jpcadena

app/config/exceptions.py

@@ -0,0 +1,16 @@
+"""
+This module defines custom exception classes for the Core
+"""
+
+from sqlalchemy.exc import SQLAlchemyError
+
+
+class DatabaseException(SQLAlchemyError):  # type: ignore
+    """
+    Database Exception class
+    """
+
+    def __init__(self, message: str, note: str | None = None):
+        super().__init__(message)
+        if note:
+            self.add_note(note)

app/crud/user.py

@@ -0,0 +1,185 @@
+"""
+This script handles CRUD (Create, Read, Update, Delete) operations for
+ User objects in the database.
+"""
+
+import logging
+from datetime import UTC, datetime
+from typing import Any, Sequence
+
+from pydantic import NonNegativeInt, PositiveInt
+from sqlalchemy import Row, RowMapping, select
+from sqlalchemy.engine import ScalarResult
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.orm import Session
+from sqlalchemy.sql import Select
+
+from app.config.exceptions import DatabaseException
+from app.db.session import get_session
+from app.models.user import User
+from app.schemas.user import UserCreate, UserUpdate
+
+logger: logging.Logger = logging.getLogger(__name__)
+
+
+class UserRepository:
+    """
+    This class handles all operations (CRUD) related to a User in the
+     database.
+    """
+
+    def __init__(
+        self,
+        session: Session,
+    ):
+        self.session: Session = session
+
+    def read_by_id(self, _id: PositiveInt) -> User | None:
+        """
+        Retrieve a user from the database by its id
+        :param _id: The id of the user
+        :type _id: IdSpecification
+        :return: The user with the specified id, or None if no such
+            user exists
+        :rtype: Optional[User]
+        """
+        with self.session as session:
+            stmt: Select[Any]
+            stmt = select(User).where(User.id == _id)
+            try:
+                db_obj: Row[Any] | RowMapping = (session.scalars(stmt)).one()
+                if not isinstance(db_obj, User):
+                    raise ValueError("Retrieved object is not a User instance")
+            except SQLAlchemyError as sa_exc:
+                logger.error(sa_exc)
+                logger.info("Retrieving row with id: %s", _id)
+                raise DatabaseException(str(sa_exc)) from sa_exc
+            return User(db_obj)
+
+    def read_users(
+        self,
+        offset: NonNegativeInt,
+        limit: PositiveInt,
+    ) -> list[User]:
+        """
+        Retrieve a list of users from the database, with pagination
+        :param offset: The number of users to skip before starting to
+         return users
+        :type offset: NonNegativeInt
+        :param limit: The maximum number of users to return
+        :type limit: PositiveInt
+        :return: A list of users
+        :rtype: list[User]
+        """
+        stmt: Select[tuple[User]] = select(User).offset(offset).limit(limit)
+        with self.session as session:
+            try:
+                scalar_result: ScalarResult[User] = session.scalars(stmt)
+                all_results: Sequence[Row[User] | RowMapping | Any] = (
+                    scalar_result.all()
+                )
+                users: list[User] = [User(result) for result in all_results]
+            except SQLAlchemyError as sa_exc:
+                logger.error(sa_exc)
+                raise DatabaseException(str(sa_exc)) from sa_exc
+            return users
+
+    def create_user(
+        self,
+        user: UserCreate,
+    ) -> User:
+        """
+        Create a new user in the database.
+        :param user: An object containing the information of the user
+         to create
+        :type user: UserCreate
+        :return: The created user object
+        :rtype: User
+        """
+        user_data: dict[str, Any] = user.model_dump()
+        user_create: User = User(**user_data)
+        with self.session as session:
+            try:
+                session.add(user_create)
+                session.commit()
+            except SQLAlchemyError as sa_exc:
+                logger.error(sa_exc)
+                session.rollback()
+                raise DatabaseException(str(sa_exc)) from sa_exc
+            if created_user := self.read_by_id(user_create.id):
+                return created_user
+            else:
+                raise DatabaseException("User could not be created")
+
+    def update_user(self, _id: PositiveInt, user: UserUpdate) -> User | None:
+        """
+        Update the information of a user in the database
+        :param _id: The id of the user to update
+        :type _id: PositiveInt
+        :param user: An object containing the new information of the
+         user
+        :type user: UserUpdate
+        :return: The updated user, or None if no such user exists
+        :rtype: Optional[User]
+        """
+        with self.session as session:
+            try:
+                found_user: User | None = self.read_by_id(_id)
+            except DatabaseException as db_exc:
+                logger.error(db_exc)
+                raise DatabaseException(str(db_exc)) from db_exc
+            if not found_user:
+                raise DatabaseException(
+                    f"User with ID: {_id} could not be updated"
+                )
+            update_data: dict[str, Any] = user.model_dump(exclude_unset=True)
+            for field, value in update_data.items():
+                if value is not None:
+                    setattr(found_user, field, value)
+            found_user.updated_at = datetime.now(UTC)
+            session.add(found_user)
+            session.commit()
+            try:
+                updated_user: User | None = self.read_by_id(_id)
+            except DatabaseException as db_exc:
+                logger.error(db_exc)
+                raise DatabaseException(str(db_exc)) from db_exc
+            return updated_user
+
+    def delete_user(self, _id: PositiveInt) -> bool:
+        """
+        Delete a user from the database
+        :param _id: The id of the user to delete
+        :type _id: PositiveInt
+        :return: True if the user is deleted; otherwise False
+        :rtype: bool
+        """
+        with self.session as session:
+            try:
+                found_user: User | None = self.read_by_id(_id)
+            except DatabaseException as db_exc:
+                raise DatabaseException(str(db_exc)) from db_exc
+            if not found_user:
+                raise DatabaseException(
+                    f"User with ID: {_id} could not be deleted"
+                )
+            try:
+                session.delete(found_user)
+                session.commit()
+            except SQLAlchemyError as sa_exc:
+                logger.error(sa_exc)
+                session.rollback()
+                return False
+            return True
+
+
+def get_user_repository() -> UserRepository:
+    """
+    Create a UserRepository with a database session, an index
+     filter, and a unique filter.
+    :return: A UserRepository instance
+    :rtype: UserRepository
+    """
+    return UserRepository(
+        get_session(),
+    )

app/db/session.py

@@ -11,7 +11,7 @@
 engine: Engine = create_engine(url, pool_pre_ping=True, future=True, echo=True)
 
 
-async def get_session() -> Session:
+def get_session() -> Session:
     """
     Get an asynchronous session to the database
     :return session: Session for database connection

app/models/user.py

@@ -0,0 +1,83 @@
+"""
+A module for user in the app models package.
+"""
+
+from datetime import datetime
+
+from pydantic import EmailStr, PastDate, PositiveInt
+from pydantic_extra_types.phone_numbers import PhoneNumber
+from sqlalchemy import CheckConstraint, Date, Integer, String, text
+from sqlalchemy.dialects.postgresql import TIMESTAMP
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.config.settings import setting
+from app.db.base import Base
+
+
+class User(Base):  # type: ignore
+    """
+    User model class representing the "users" table
+    """
+
+    __tablename__ = "users"
+
+    id: Mapped[PositiveInt] = mapped_column(
+        Integer,
+        nullable=False,
+        primary_key=True,
+        autoincrement=True,
+        index=True,
+        unique=True,
+        comment="ID of the User",
+    )
+    username: Mapped[str] = mapped_column(
+        String(15),
+        index=True,
+        unique=True,
+        nullable=False,
+        comment="Username to identify the user",
+    )
+    email: Mapped[EmailStr] = mapped_column(
+        String(320),
+        index=True,
+        unique=True,
+        nullable=False,
+        comment="Preferred e-mail address of the User",
+    )
+    password: Mapped[str] = mapped_column(
+        String(60), nullable=False, comment="Hashed password of the User"
+    )
+    birthdate: Mapped[PastDate] = mapped_column(
+        Date, nullable=True, comment="Birthday of the User"
+    )
+    phone_number: Mapped[PhoneNumber] = mapped_column(
+        String(20),
+        nullable=True,
+        comment="Preferred telephone number of the User",
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        TIMESTAMP(timezone=True, precision=setting.TIMESTAMP_PRECISION),
+        default=datetime.now(),
+        nullable=False,
+        server_default=text("now()"),
+        comment="Time the User was created",
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        TIMESTAMP(timezone=True, precision=setting.TIMESTAMP_PRECISION),
+        nullable=True,
+        onupdate=text("now()"),
+        comment="Time the User was updated",
+    )
+
+    __table_args__ = (
+        CheckConstraint(
+            "char_length(username) >= 4", name="users_username_length"
+        ),
+        CheckConstraint("char_length(email) >= 3", name="users_email_length"),
+        CheckConstraint(setting.DB_EMAIL_CONSTRAINT, name="users_email_format"),
+        CheckConstraint("LENGTH(password) = 60", name="users_password_length"),
+        CheckConstraint(
+            setting.DB_PHONE_NUMBER_CONSTRAINT,
+            name="users_phone_number_format",
+        ),
+    )

app/schemas/user.py

@@ -0,0 +1,13 @@
+"""
+A module for user in the app-schemas package.
+"""
+
+from pydantic import BaseModel
+
+
+class UserCreate(BaseModel):
+    pass
+
+
+class UserUpdate(BaseModel):
+    pass