Merge pull request #10 from e-gulakhmet/HF-43

HF-43: Implemented ValueObjects for Entities and InputDTOs

src/expenses/__init__.py

@@ -0,0 +1,20 @@
+from .entities import Expense
+from .exceptions import UserShouldHavePrimaryStorageError
+from .types import Amount, Category, ExpensesStorageLink, OwnerID
+from .usecases import ExpenseCreateInput, ExpenseCreateRepoInterface, ExpenseCreateUseCase
+
+__all__ = [
+    # entities
+    "Expense",
+    # exceptions
+    "UserShouldHavePrimaryStorageError",
+    # types
+    "Amount",
+    "Category",
+    "ExpensesStorageLink",
+    "OwnerID",
+    # usecases
+    "ExpenseCreateInput",
+    "ExpenseCreateRepoInterface",
+    "ExpenseCreateUseCase",
+]

src/expenses/entities/expense.py

@@ -1,13 +1,14 @@
-import uuid
 from dataclasses import dataclass, field
 from datetime import datetime
 
+from src.expenses.types import Amount, Category, ExpensesStorageLink, OwnerID
+
 
 @dataclass
 class Expense:
-    user_id: uuid.UUID
-    exepenses_storage_link: str
-    amount: float
-    category: str
+    owner_id: OwnerID
+    expenses_storage_link: ExpensesStorageLink
+    amount: Amount
+    category: Category
     subcategory: str = ""
     created_at: datetime = field(default_factory=datetime.now)

src/expenses/types.py

@@ -0,0 +1,41 @@
+import re
+from typing import NewType
+from uuid import UUID
+
+OwnerID = NewType("OwnerID", UUID)
+
+
+class ExpensesStorageLink(str):
+    __slots__ = ()
+
+    _GOOGLE_SHEETS_LINK_PATTERN = r"https:\/\/docs\.google\.com\/spreadsheets\/d\/[a-zA-Z0-9_-]+"
+    _INVALID_LINK_ERROR_TEXT = "Expenses Storage link is invalid"
+
+    def __new__(cls, value: str) -> "ExpensesStorageLink":
+        if not re.match(pattern=cls._GOOGLE_SHEETS_LINK_PATTERN, string=value):
+            raise ValueError(cls._INVALID_LINK_ERROR_TEXT)
+        return super().__new__(cls, value)
+
+
+class Amount(float):
+    _AMOUNT_MIN_VALUE_ERROR_MESSAGE = "Amount cannot be negative"
+
+    def __new__(cls, value: float) -> "Amount":
+        if value <= 0:
+            raise ValueError(cls._AMOUNT_MIN_VALUE_ERROR_MESSAGE)
+        return super().__new__(cls, value)
+
+
+class Category(str):
+    __slots__ = ()
+
+    _CATEGORY_MAX_LENGTH = 50
+    _CATEGORY_MIN_LENGTH = 3
+    _CATEGORY_LENGTH_ERROR_MESSAGE = (
+        f"Category must be between {_CATEGORY_MIN_LENGTH} and {_CATEGORY_MAX_LENGTH} characters"
+    )
+
+    def __new__(cls, value: str) -> "Category":
+        if not cls._CATEGORY_MIN_LENGTH <= len(value) <= cls._CATEGORY_MAX_LENGTH:
+            raise ValueError(cls._CATEGORY_LENGTH_ERROR_MESSAGE)
+        return super().__new__(cls, value)

src/expenses/usecases/create.py

@@ -1,12 +1,12 @@
 import abc
-import uuid
 from dataclasses import dataclass, field
 from datetime import datetime
 
 from src.expenses.entities import Expense
 from src.expenses.exceptions import UserShouldHavePrimaryStorageError
-from src.storages.entities import Storage
-from src.storages.usecases import StorageGetUseCase
+from src.expenses.types import Amount, Category, ExpensesStorageLink, OwnerID
+from src.storages import OwnerID as StorageOwnerID
+from src.storages import StorageGetUseCase
 
 
 class ExpenseCreateRepoInterface(abc.ABC):
@@ -17,9 +17,9 @@ async def save(self, expense: Expense) -> None:
 
 @dataclass
 class ExpenseCreateInput:
-    user_id: uuid.UUID
-    amount: float
-    category: str
+    owner_id: OwnerID
+    amount: Amount
+    category: Category
     subcategory: str = ""
     created_at: datetime = field(default_factory=datetime.now)
 
@@ -35,24 +35,21 @@ def __init__(
 
     async def execute(self, input_: ExpenseCreateInput) -> Expense:
         # Getting User's Primary Storage
-        storage = await self._storage_get_usecase.execute(filter_={"user_id": input_.user_id, "primary": True})
+        storage = await self._storage_get_usecase.execute(
+            filter_={"owner_id": StorageOwnerID(input_.owner_id), "primary": True},
+        )
         if not storage:
             raise UserShouldHavePrimaryStorageError
 
-        # Creating domain entity
-        expense = self._create_domain(input_=input_, storage=storage)
-
-        # Saving Storage to repository
-        await self._expense_repo.save(expense=expense)
-
-        return expense
-
-    def _create_domain(self, input_: ExpenseCreateInput, storage: Storage) -> Expense:
-        return Expense(
-            user_id=input_.user_id,
-            exepenses_storage_link=storage.expenses_table_link,
+        expense = Expense(
+            owner_id=input_.owner_id,
+            expenses_storage_link=ExpensesStorageLink(storage.expenses_table_link),
             amount=input_.amount,
             category=input_.category,
             subcategory=input_.subcategory,
             created_at=input_.created_at,
         )
+
+        await self._expense_repo.save(expense)
+
+        return expense

src/storages/__init__.py

@@ -0,0 +1,26 @@
+from .entities import Storage
+from .types import OwnerID, StorageID, StorageLink
+from .usecases import (
+    StorageCreateInput,
+    StorageCreateRepoInterface,
+    StorageCreateUseCase,
+    StorageGetFilter,
+    StorageGetRepoInterface,
+    StorageGetUseCase,
+)
+
+__all__ = [
+    # entities
+    "Storage",
+    # types
+    "OwnerID",
+    "StorageID",
+    "StorageLink",
+    # usecases
+    "StorageCreateInput",
+    "StorageCreateRepoInterface",
+    "StorageCreateUseCase",
+    "StorageGetFilter",
+    "StorageGetRepoInterface",
+    "StorageGetUseCase",
+]

src/storages/entities/storage.py

@@ -2,14 +2,16 @@
 import uuid
 from dataclasses import dataclass, field
 
+from src.storages.types import OwnerID, StorageID, StorageLink
+
 
 @dataclass
 class Storage:
-    link: str
-    expenses_table_link: str
-    income_table_link: str
-    user_id: uuid.UUID
+    link: StorageLink
+    expenses_table_link: StorageLink
+    income_table_link: StorageLink
+    owner_id: OwnerID
     primary: bool = False
 
-    id: uuid.UUID = field(default_factory=uuid.uuid4)
+    id: StorageID = field(default_factory=lambda: StorageID(uuid.uuid4()))
     created_at: datetime.datetime = field(default_factory=datetime.datetime.now)

src/storages/types.py

@@ -0,0 +1,20 @@
+import re
+from typing import NewType
+from uuid import UUID
+
+
+class StorageLink(str):
+    __slots__ = ()
+
+    _GOOGLE_SHEETS_LINK_PATTERN = r"https:\/\/docs\.google\.com\/spreadsheets\/d\/[a-zA-Z0-9_-]+"
+    _INVALID_LINK_ERROR_TEXT = "Storage link is invalid"
+
+    def __new__(cls, value: str) -> "StorageLink":
+        if not re.match(pattern=cls._GOOGLE_SHEETS_LINK_PATTERN, string=value):
+            raise ValueError(cls._INVALID_LINK_ERROR_TEXT)
+        return super().__new__(cls, value)
+
+
+StorageID = NewType("StorageID", UUID)
+
+OwnerID = NewType("OwnerID", UUID)

src/storages/usecases/__init__.py

@@ -4,6 +4,7 @@
     StorageCreateUseCase,
 )
 from .get import (
+    StorageGetFilter,
     StorageGetRepoInterface,
     StorageGetUseCase,
 )
@@ -14,4 +15,5 @@
     "StorageCreateUseCase",
     "StorageGetRepoInterface",
     "StorageGetUseCase",
+    "StorageGetFilter",
 ]

src/storages/usecases/create.py

@@ -1,10 +1,9 @@
 import abc
-import re
-import uuid
 from dataclasses import dataclass
 
 from src.exceptions import ValidationError
 from src.storages.entities import Storage
+from src.storages.types import OwnerID, StorageLink
 
 
 class StorageCreateRepoInterface(abc.ABC):
@@ -13,80 +12,62 @@ async def save(self, storage: Storage) -> None:
         ...
 
     @abc.abstractmethod
-    async def exists(self, user_id: uuid.UUID, primary: bool) -> bool:
+    async def exists(self, owner_id: OwnerID) -> bool:
         ...
 
     @abc.abstractmethod
-    async def is_accessable(self, link: str) -> bool:
+    async def is_accessable(self, link: StorageLink) -> bool:
         ...
 
 
 @dataclass
 class StorageCreateInput:
-    link: str
-    expenses_table_link: str
-    income_table_link: str
-    user_id: uuid.UUID
+    link: StorageLink
+    expenses_table_link: StorageLink
+    income_table_link: StorageLink
+    owner_id: OwnerID
 
 
 class StorageCreateUseCase:
+    _STORAGE_LINK_IS_NOT_ACCESSABLE_ERROR_TEXT = "Storage link is not accessable"
+    _EXPENSES_TABLE_LINK_IS_NOT_ACCESSABLE_ERROR_TEXT = "Expenses table link is not accessable"
+    _INCOME_TABLE_LINK_IS_NOT_ACCESSABLE_ERROR_TEXT = "Income table link is not accessable"
+
     def __init__(
         self,
         storage_repo: StorageCreateRepoInterface,
     ) -> None:
         self._storage_repo = storage_repo
 
     async def execute(self, input_: StorageCreateInput) -> Storage:
-        # Validating input
         await self._validate(input_=input_)
 
-        # Creating domain entity
-        storage = self._create_domain(input_=input_)
+        storage = Storage(
+            link=input_.link,
+            expenses_table_link=input_.expenses_table_link,
+            income_table_link=input_.income_table_link,
+            owner_id=input_.owner_id,
+        )
 
-        # Setting primary flag to Storage
-        # If User has no primary storage, then set primary flag to True
-        storage.primary = bool(not await self._storage_repo.exists(user_id=input_.user_id, primary=True))
+        if not await self._storage_repo.exists(owner_id=input_.owner_id):
+            storage.primary = True
 
-        # Saving Storage to repository
         await self._storage_repo.save(storage=storage)
 
         return storage
 
     async def _validate(self, input_: StorageCreateInput) -> None:
-        await self._validate_storage_link(link=input_.link)
-        await self._validate_expenses_table_link(link=input_.expenses_table_link)
-        await self._validate_income_table_link(link=input_.income_table_link)
-
-    async def _validate_storage_link(self, link: str) -> None:
-        if not self._is_correct_storage_link_format(link=link):
-            raise ValidationError(field="link", message="Invalid link to Storage")
-
-        if not await self._storage_repo.is_accessable(link=link):
-            raise ValidationError(field="link", message="Storage is not accessable")
-
-    async def _validate_expenses_table_link(self, link: str) -> None:
-        if not self._is_correct_storage_link_format(link=link):
-            raise ValidationError(field="expenses_table_link", message="Invalid link to Expenses table")
-
-        if not await self._storage_repo.is_accessable(link=link):
-            raise ValidationError(field="expenses_table_link", message="Expenses table is not accessable")
-
-    async def _validate_income_table_link(self, link: str) -> None:
-        if not self._is_correct_storage_link_format(link=link):
-            raise ValidationError(field="income_table_link", message="Invalid link to Income table")
-
-        if not await self._storage_repo.is_accessable(link=link):
-            raise ValidationError(field="income_table_link", message="Income table is not accessable")
-
-    def _is_correct_storage_link_format(self, link: str) -> bool:
-        # Check if link is Google Sheets link
-        pattern = r"https:\/\/docs\.google\.com\/spreadsheets\/d\/[a-zA-Z0-9_-]+"
-        return re.match(pattern=pattern, string=link) is not None
-
-    def _create_domain(self, input_: StorageCreateInput) -> Storage:
-        return Storage(
-            link=input_.link,
-            expenses_table_link=input_.expenses_table_link,
-            income_table_link=input_.income_table_link,
-            user_id=input_.user_id,
-        )
+        if not await self._storage_repo.is_accessable(link=input_.link):
+            raise ValidationError(field="link", message=self._STORAGE_LINK_IS_NOT_ACCESSABLE_ERROR_TEXT)
+
+        if not await self._storage_repo.is_accessable(link=input_.expenses_table_link):
+            raise ValidationError(
+                field="expenses_table_link",
+                message=self._EXPENSES_TABLE_LINK_IS_NOT_ACCESSABLE_ERROR_TEXT,
+            )
+
+        if not await self._storage_repo.is_accessable(link=input_.income_table_link):
+            raise ValidationError(
+                field="income_table_link",
+                message=self._INCOME_TABLE_LINK_IS_NOT_ACCESSABLE_ERROR_TEXT,
+            )

src/storages/usecases/get.py

@@ -1,26 +1,26 @@
 import abc
-import uuid
 from typing import TypedDict
 
 from src.storages.entities import Storage
+from src.storages.types import OwnerID, StorageID, StorageLink
 
 
-class Filter(TypedDict, total=False):
-    id: uuid.UUID
-    user_id: uuid.UUID
-    link: str
+class StorageGetFilter(TypedDict, total=False):
+    id: StorageID
+    owner_id: OwnerID
+    link: StorageLink
     primary: bool
 
 
 class StorageGetRepoInterface(abc.ABC):
     @abc.abstractmethod
-    async def get(self, filter_: Filter) -> Storage:
+    async def get(self, filter_: StorageGetFilter) -> Storage:
         ...
 
 
 class StorageGetUseCase:
     def __init__(self, storage_repo: StorageGetRepoInterface) -> None:
         self._storage_repo = storage_repo
 
-    async def execute(self, filter_: Filter) -> Storage:
+    async def execute(self, filter_: StorageGetFilter) -> Storage:
         return await self._storage_repo.get(filter_=filter_)