Merge pull request #1 from Roman505050/feature/transactions

Feature/transactions

5000

@@ -0,0 +1 @@
+13503

pyproject.toml

@@ -17,6 +17,9 @@ asyncpg = "^0.30.0"
 python-dotenv = "^1.0.1"
 gunicorn = "^23.0.0"
 loguru = "^0.7.2"
+flasgger = "^0.9.7.1"
+asgiref = "^3.8.1"
+hypercorn = "^0.17.3"
 
 
 [tool.poetry.group.dev.dependencies]

requirements.txt

@@ -1,25 +1,41 @@
 annotated-types==0.7.0
 asgiref==3.8.1
 asyncpg==0.30.0
+attrs==24.2.0
 bcrypt==4.2.0
 blinker==1.8.2
 click==8.1.7
 dnspython==2.7.0
 email_validator==2.2.0
+flasgger==0.9.7.1
 Flask==3.0.3
 Flask-WTF==1.2.2
 greenlet==3.1.1
 gunicorn==23.0.0
+h11==0.14.0
+h2==4.1.0
+hpack==4.0.0
+Hypercorn==0.17.3
+hyperframe==6.0.1
 idna==3.10
 itsdangerous==2.2.0
 Jinja2==3.1.4
+jsonschema==4.23.0
+jsonschema-specifications==2024.10.1
+loguru==0.7.2
 MarkupSafe==3.0.2
+mistune==3.0.2
 packaging==24.1
+priority==2.0.0
 pydantic==2.9.2
 pydantic_core==2.23.4
 python-dotenv==1.0.1
+PyYAML==6.0.2
+referencing==0.35.1
+rpds-py==0.21.0
+six==1.16.0
 SQLAlchemy==2.0.36
 typing_extensions==4.12.2
 Werkzeug==3.1.2
+wsproto==1.2.0
 WTForms==3.2.1
-loguru~=0.7.2

src/core/application/transaction/dto/category.py

@@ -0,0 +1,23 @@
+from pydantic import BaseModel, Field
+from uuid import UUID
+
+from core.domain.transaction.entities.category import CategoryEntity
+
+
+class CreateCategoryDTO(BaseModel):
+    category_name: str = Field(min_length=3, max_length=64)
+    operation_id: UUID
+
+
+class CategoryDTO(BaseModel):
+    category_id: UUID
+    category_name: str
+    operation_name: str
+
+    @staticmethod
+    def from_entity(entity: CategoryEntity) -> "CategoryDTO":
+        return CategoryDTO(
+            category_id=entity.category_id,
+            category_name=entity.category_name,
+            operation_name=entity.operation.operation_name,
+        )

src/core/application/transaction/dto/currency.py

@@ -0,0 +1,39 @@
+from pydantic import BaseModel, Field, StringConstraints
+from pydantic_core import PydanticCustomError
+from typing import Annotated
+from uuid import UUID
+
+from core.domain.transaction.entities.currency import CurrencyEntity
+from core.shared.utils import custom_error_msg
+
+
+CurrencyCode = Annotated[
+    str,
+    StringConstraints(pattern=r"^[A-Z]{3}$", min_length=3, max_length=3),
+    custom_error_msg(
+        lambda _, __: PydanticCustomError(
+            "str_error",
+            "Currency code should consist of "
+            "exactly 3 uppercase letters (e.g., USD, EUR).",
+        )
+    ),
+]
+
+
+class CreateCurrencyDTO(BaseModel):
+    currency_code: CurrencyCode
+    currency_name: str = Field(min_length=3, max_length=64)
+    currency_symbol: str = Field(min_length=1, max_length=8)
+
+
+class CurrencyDTO(CreateCurrencyDTO):
+    currency_id: UUID
+
+    @staticmethod
+    def from_entity(entity: CurrencyEntity) -> "CurrencyDTO":
+        return CurrencyDTO(
+            currency_id=entity.currency_id,
+            currency_code=entity.currency_code,
+            currency_name=entity.currency_name,
+            currency_symbol=entity.currency_symbol,
+        )

src/core/application/transaction/dto/operation.py

@@ -0,0 +1,25 @@
+from pydantic import BaseModel, Field
+from typing import Literal
+from uuid import UUID
+
+from core.domain.transaction.entities.operation import OperationEntity
+from core.domain.transaction.enums.operation import (
+    get_operation_type_string,
+)
+
+
+class CreateOperationDTO(BaseModel):
+    operation_name: str = Field(min_length=3, max_length=64)
+    operation_type: Literal["income", "expense", "investment"]
+
+
+class OperationDTO(CreateOperationDTO):
+    operation_id: UUID
+
+    @staticmethod
+    def from_entity(entity: OperationEntity) -> "OperationDTO":
+        return OperationDTO(
+            operation_id=entity.operation_id,
+            operation_name=entity.operation_name,
+            operation_type=get_operation_type_string(entity.operation_type),
+        )

src/core/application/transaction/dto/transaction.py

@@ -0,0 +1,57 @@
+from pydantic import BaseModel, Field
+from uuid import UUID
+from decimal import Decimal
+from typing import Literal
+import datetime
+
+from core.domain.transaction.entities.transaction import TransactionEntity
+from core.domain.transaction.enums.operation import (
+    get_operation_type_string,
+)
+
+
+class CreateTransactionDTO(BaseModel):
+    user_id: UUID
+    category_id: UUID
+    currency_id: UUID
+    amount: Decimal = Field(gt=0, le=Decimal("99999999"))
+    description: str | None = Field(min_length=10, max_length=255)
+    date: datetime.datetime
+
+
+class TransactionDTO(BaseModel):
+    transaction_id: UUID
+    user_id: UUID
+    category_id: UUID
+    category_name: str
+    operation_id: UUID
+    operation_name: str
+    operation_type: Literal["income", "expense", "investment"]
+    currency_id: UUID
+    currency_name: str
+    currency_code: str
+    currency_symbol: str
+    amount: Decimal
+    description: str | None
+    date: datetime.datetime
+
+    @staticmethod
+    def from_entity(entity: TransactionEntity) -> "TransactionDTO":
+        return TransactionDTO(
+            transaction_id=entity.transaction_id,
+            user_id=entity.user_id,
+            category_id=entity.category.category_id,
+            category_name=entity.category.category_name,
+            operation_id=entity.category.operation.operation_id,
+            operation_name=entity.category.operation.operation_name,
+            operation_type=get_operation_type_string(
+                entity.category.operation.operation_type
+            ),
+            currency_id=entity.money.currency.currency_id,
+            currency_name=entity.money.currency.currency_name,
+            currency_code=entity.money.currency.currency_code,
+            currency_symbol=entity.money.currency.currency_symbol,
+            amount=entity.money.amount,
+            description=entity.description,
+            date=entity.date,
+        )

src/core/application/transaction/factories/category.py

@@ -0,0 +1,16 @@
+from uuid import uuid4
+
+from core.domain.transaction.entities.category import CategoryEntity
+from core.domain.transaction.entities.operation import OperationEntity
+
+
+class CategoryFactory:
+    @staticmethod
+    def create(
+        category_name: str, operation: OperationEntity
+    ) -> CategoryEntity:
+        return CategoryEntity(
+            category_id=uuid4(),
+            category_name=category_name,
+            operation=operation,
+        )

src/core/application/transaction/factories/currency.py

@@ -0,0 +1,16 @@
+from uuid import uuid4
+
+from core.domain.transaction.entities.currency import CurrencyEntity
+
+
+class CurrencyFactory:
+    @staticmethod
+    def create(
+        currency_name: str, currency_code: str, currency_symbol: str
+    ) -> CurrencyEntity:
+        return CurrencyEntity(
+            currency_id=uuid4(),
+            currency_name=currency_name,
+            currency_code=currency_code,
+            currency_symbol=currency_symbol,
+        )

src/core/application/transaction/factories/operation.py

@@ -0,0 +1,18 @@
+from uuid import uuid4
+from typing import Literal
+
+from core.domain.transaction.entities.operation import OperationEntity
+from core.domain.transaction.enums.operation import get_operation_type_enum
+
+
+class OperationFactory:
+    @staticmethod
+    def create(
+        operation_name: str,
+        operation_type: Literal["income", "expense", "investment"],
+    ) -> OperationEntity:
+        return OperationEntity(
+            operation_id=uuid4(),
+            operation_name=operation_name,
+            operation_type=get_operation_type_enum(operation_type),
+        )

src/core/application/transaction/factories/transaction.py

@@ -0,0 +1,31 @@
+from decimal import Decimal
+from uuid import uuid4, UUID
+import datetime
+
+from core.domain.transaction.entities.category import CategoryEntity
+from core.domain.transaction.entities.currency import CurrencyEntity
+from core.domain.transaction.entities.transaction import TransactionEntity
+from core.domain.transaction.value_objects.money import Money
+
+
+class TransactionFactory:
+    @staticmethod
+    def create(
+        user_id: UUID,
+        category: CategoryEntity,
+        currency: CurrencyEntity,
+        amount: Decimal,
+        description: str | None,
+        date: datetime.datetime,
+    ) -> TransactionEntity:
+        return TransactionEntity(
+            transaction_id=uuid4(),
+            user_id=user_id,
+            category=category,
+            money=Money(
+                currency=currency,
+                amount=amount,
+            ),
+            description=description,
+            date=date,
+        )

src/core/application/transaction/use_cases/category/create.py

@@ -0,0 +1,38 @@
+from core.application.transaction.factories.category import CategoryFactory
+from core.domain.transaction.repositories.category import ICategoryRepository
+from core.application.transaction.dto.category import (
+    CategoryDTO,
+    CreateCategoryDTO,
+)
+from core.domain.transaction.repositories.operation import IOperationRepository
+
+
+class CreateCategoryUseCase:
+    def __init__(
+        self,
+        category_repository: ICategoryRepository,
+        operation_repository: IOperationRepository,
+    ):
+        self._category_repository = category_repository
+        self._operation_repository = operation_repository
+
+    async def execute(self, request: CreateCategoryDTO) -> CategoryDTO:
+        """
+        Create a operation.
+
+        :arg request: The request data.
+        :raise OperationNotFoundException: If the operation does not exist.
+        :raise CategoryAlreadyExistsException: If the operation already exists.
+        :return: The created operation.
+        """
+        operation = await self._operation_repository.get_by_id(
+            request.operation_id
+        )
+
+        category = CategoryFactory.create(
+            category_name=request.category_name, operation=operation
+        )
+
+        category = await self._category_repository.save(category)
+
+        return CategoryDTO.from_entity(category)

src/core/application/transaction/use_cases/category/delete.py

@@ -0,0 +1,19 @@
+from uuid import UUID
+
+from core.domain.transaction.repositories.category import ICategoryRepository
+
+
+class DeleteCategoryUseCase:
+    def __init__(self, category_repository: ICategoryRepository):
+        self._category_repository = category_repository
+
+    async def execute(self, category_id: UUID) -> None:
+        """Delete a operation.
+
+        :arg category_id: The operation ID.
+        :raise CategoryNotFoundException: If the operation does not exist.
+        :raise CategoryNotDeletableException: If the operation
+                                                is not deletable.
+        :return: None
+        """
+        await self._category_repository.delete(category_id)

src/core/application/transaction/use_cases/category/get_all.py

@@ -0,0 +1,15 @@
+from core.application.transaction.dto.category import CategoryDTO
+from core.domain.transaction.repositories.category import ICategoryRepository
+
+
+class GetAllCategoriesUseCase:
+    def __init__(self, category_repository: ICategoryRepository):
+        self._category_repository = category_repository
+
+    async def execute(self) -> list[CategoryDTO]:
+        """Get all categories.
+
+        :return: The categories.
+        """
+        entities = await self._category_repository.get_all()
+        return [CategoryDTO.from_entity(entity) for entity in entities]

src/core/application/transaction/use_cases/category/get_all_by_operation.py

@@ -0,0 +1,19 @@
+from uuid import UUID
+
+from core.application.transaction.dto.category import CategoryDTO
+from core.domain.transaction.repositories.category import ICategoryRepository
+
+
+class GetAllCategoriesByOperationUseCase:
+    def __init__(self, category_repository: ICategoryRepository):
+        self._category_repository = category_repository
+
+    async def execute(self, operation_id: UUID) -> list[CategoryDTO]:
+        """Get all categories by operation.
+
+        :return The categories.
+        """
+        entities = await self._category_repository.get_by_operation_id(
+            operation_id=operation_id
+        )
+        return [CategoryDTO.from_entity(entity) for entity in entities]