feat: TT-418 create added

V2/serverless.yml

@@ -96,3 +96,13 @@ functions:
             - DELETE
           route: time-entries/{id}
           authLevel: anonymous
+
+  create_customer:
+    handler: time_tracker/customers/interface.create_customer
+    events:
+      - http: true
+        x-azure-settings:
+          methods:
+            - POST
+          route: customer/
+          authLevel: anonymous

V2/time_tracker/customers/_application/__init__.py

@@ -0,0 +1,2 @@
+# flake8: noqa
+from ._time_entries import create_customer

V2/time_tracker/customers/_application/_time_entries/__init__.py

@@ -0,0 +1,2 @@
+# flake8: noqa
+from ._create_customer import create_customer

V2/time_tracker/customers/_application/_time_entries/_create_customer.py

@@ -0,0 +1,56 @@
+import dataclasses
+import json
+import typing
+
+import azure.functions as func
+
+from ... import _domain
+from ... import _infrastructure
+from time_tracker._infrastructure import DB
+
+DEFAULT_FIELDS = ["id", "deleted", "status"]
+
+
+def create_customer(req: func.HttpRequest) -> func.HttpResponse:
+    database = DB()
+    customer_dao = _infrastructure.CustomersSQLDao(database)
+    customer_service = _domain.CustomerService(customer_dao)
+    use_case = _domain._use_cases.CreateCustomerUseCase(customer_service)
+
+    customer_data = req.get_json()
+    validation_errors = _validate_customer(customer_data)
+    if validation_errors:
+        return func.HttpResponse(
+            body=validation_errors, status_code=400, mimetype="application/json"
+        )
+
+    customer_to_create = _domain.Customer(
+        id=None,
+        name=customer_data["name"],
+        description=customer_data["description"],
+        deleted=False,
+        status=-1
+    )
+
+    created_customer = use_case.create_customer(customer_to_create)
+
+    if not created_customer:
+        return func.HttpResponse(
+            body=json.dumps({'error': 'customer could not be created'}),
+            status_code=500,
+            mimetype="application/json"
+        )
+
+    return func.HttpResponse(
+        body=json.dumps(created_customer.__dict__),
+        status_code=201,
+        mimetype="application/json"
+    )
+
+
+def _validate_customer(customer_data: dict) -> typing.List[str]:
+    customer_fields = [field.name for field in dataclasses.fields(_domain.Customer) 
+                        if field.name not in DEFAULT_FIELDS]
+    if [field for field in customer_fields if field not in customer_data]:
+        return b'Invalid format or structure of the attributes of the customer'
+    return []

V2/time_tracker/customers/_application/_time_entries/_delete_time_entry.py

@@ -0,0 +1,36 @@
+import json
+
+import azure.functions as func
+
+from ... import _domain
+from ... import _infrastructure
+from time_tracker._infrastructure import DB
+
+
+def delete_time_entry(req: func.HttpRequest) -> func.HttpResponse:
+    time_entry_dao = _infrastructure.TimeEntriesSQLDao(DB())
+    time_entry_service = _domain.TimeEntryService(time_entry_dao)
+    use_case = _domain._use_cases.DeleteTimeEntryUseCase(time_entry_service)
+
+    try:
+        time_entry_id = int(req.route_params.get("id"))
+        deleted_time_entry = use_case.delete_time_entry(time_entry_id)
+        if not deleted_time_entry:
+            return func.HttpResponse(
+                body="Not found",
+                status_code=404,
+                mimetype="application/json"
+            )
+
+        return func.HttpResponse(
+            body=json.dumps(deleted_time_entry.__dict__, default=str),
+            status_code=200,
+            mimetype="application/json",
+        )
+
+    except ValueError:
+        return func.HttpResponse(
+            body=b"Invalid Format ID",
+            status_code=400,
+            mimetype="application/json"
+        )

V2/time_tracker/customers/_domain/__init__.py

@@ -0,0 +1,11 @@
+# flake8: noqa
+from ._entities import Customer
+from ._persistence_contracts import CustomersDao
+from ._services import CustomerService
+from ._use_cases import (
+    CreateCustomerUseCase,
+    UpdateCustomerUseCase,
+    GetAllCustomerUseCase,
+    GetByIdCustomerUseCase,
+    DeleteCustomerUseCase
+)

V2/time_tracker/customers/_domain/_entities/__init__.py

@@ -0,0 +1,2 @@
+# flake8: noqa
+from ._customer import Customer

V2/time_tracker/customers/_domain/_entities/_customer.py

@@ -0,0 +1,10 @@
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class Customer:
+    id: int
+    name: str
+    description: str
+    deleted: bool
+    status: int

V2/time_tracker/customers/_domain/_persistence_contracts/__init__.py

@@ -0,0 +1,2 @@
+# flake8: noqa
+from ._customers_dao import CustomersDao

V2/time_tracker/customers/_domain/_persistence_contracts/_customers_dao.py

@@ -0,0 +1,26 @@
+import abc
+import typing
+
+from time_tracker.customers._domain import Customer
+
+
+class CustomersDao(abc.ABC):
+    @abc.abstractmethod
+    def create(self, data: Customer) -> Customer:
+        pass
+
+    # @abc.abstractmethod
+    # def update(self, data: Customer) -> Customer:
+    #     pass
+
+    # @abc.abstractmethod
+    # def get_by_id(self, id: int) -> Customer:
+    #     pass
+
+    # @abc.abstractmethod
+    # def get_all(self) -> typing.List[Customer]:
+    #     pass
+
+    # @abc.abstractmethod
+    # def delete(self, id: int) -> Customer:
+    #     pass

V2/time_tracker/customers/_domain/_services/__init__.py

@@ -0,0 +1,2 @@
+# flake8: noqa
+from ._customer import CustomerService

V2/time_tracker/customers/_domain/_services/_customer.py

@@ -0,0 +1,24 @@
+import typing
+
+from time_tracker.customers._domain import Customer, CustomersDao
+
+
+class CustomerService:
+
+    def __init__(self, customer_dao: CustomersDao):
+        self.customer_dao = customer_dao
+
+    def create(self, data: Customer) -> Customer:
+        return self.customer_dao.create(data)
+
+    # def update(self, data: Customer) -> Customer:
+    #     return self.customer_dao.update(data)
+
+    # def get_by_id(self, id: int) -> Customer:
+    #     return self.customer_dao.get_by_id(id)
+
+    # def get_all(self) -> typing.List[Customer]:
+    #     return self.customer_dao.get_all()
+
+    # def delete(self, id: int) -> Customer:
+    #     return self.customer_dao.delete(id)

V2/time_tracker/customers/_domain/_use_cases/__init__.py

@@ -0,0 +1,6 @@
+# flake8: noqa
+from ._create_customer_use_case import CreateCustomerUseCase
+from ._update_customer_use_case import UpdateCustomerUseCase
+from ._get_by_id_customer_use_case import GetByIdCustomerUseCase
+from ._get_all_customer_use_case import GetAllCustomerUseCase
+from ._delete_customer_use_case import DeleteCustomerUseCase

V2/time_tracker/customers/_domain/_use_cases/_create_customer_use_case.py

@@ -0,0 +1,10 @@
+from time_tracker.customers._domain import Customer, CustomerService
+
+
+class CreateCustomerUseCase:
+
+    def __init__(self, customer_service: CustomerService):
+        self.customer_service = customer_service
+
+    def create_customer(self, data: Customer) -> Customer:
+        return self.customer_service.create(data)

V2/time_tracker/customers/_domain/_use_cases/_delete_customer_use_case.py

@@ -0,0 +1,10 @@
+from time_tracker.customers._domain import Customer, CustomerService
+
+
+class DeleteCustomerUseCase:
+
+    def __init__(self, customer_service: CustomerService):
+        self.customer_service = customer_service
+
+    def delete_customer(self, id: int) -> Customer:
+        return self.customer_service.delete(id)

V2/time_tracker/customers/_domain/_use_cases/_get_all_customer_use_case.py

@@ -0,0 +1,12 @@
+import typing
+
+from time_tracker.customers._domain import Customer, CustomerService
+
+
+class GetAllCustomerUseCase:
+
+    def __init__(self, customer_service: CustomerService):
+        self.customer_service = customer_service
+
+    def get_all_customer(self) -> typing.List[Customer]:
+        return self.customer_service.get_all()

V2/time_tracker/customers/_domain/_use_cases/_get_by_id_customer_use_case.py

@@ -0,0 +1,10 @@
+from time_tracker.customers._domain import Customer, CustomerService
+
+
+class GetByIdCustomerUseCase:
+
+    def __init__(self, customer_service: CustomerService):
+        self.customer_service = customer_service
+
+    def get_customer_by_id(self, id: int) -> Customer:
+        return self.customer_service.get_by_id(id)

V2/time_tracker/customers/_domain/_use_cases/_update_customer_use_case.py

@@ -0,0 +1,10 @@
+from time_tracker.customers._domain import Customer, CustomerService
+
+
+class UpdateCustomerUseCase:
+
+    def __init__(self, customer_service: CustomerService):
+        self.customer_service = customer_service
+
+    def update_customer(self, data: Customer) -> Customer:
+        return self.customer_service.update(data)

V2/time_tracker/customers/_infrastructure/__init__.py

@@ -0,0 +1,2 @@
+# flake8: noqa
+from ._data_persistence import CustomersSQLDao

V2/time_tracker/customers/_infrastructure/_data_persistence/__init__.py

@@ -0,0 +1,2 @@
+# flake8: noqa
+from ._customer_dao import CustomersSQLDao

V2/time_tracker/customers/_infrastructure/_data_persistence/_customer_dao.py

@@ -0,0 +1,60 @@
+import dataclasses
+import sqlalchemy as sq
+import sqlalchemy.sql as sql
+
+import time_tracker.customers._domain as domain
+from time_tracker._infrastructure import _db
+
+
+class CustomersSQLDao(domain.CustomersDao):
+
+    def __init__(self, database: _db.DB):
+        self.customer_key = [field.name for field in dataclasses.fields(domain.Customer)]
+        self.db = database
+        self.customer = sq.Table(
+            'customer',
+            self.db.metadata,
+            sq.Column('id', sq.Integer, primary_key=True, autoincrement=True),
+            sq.Column('name', sq.String),
+            sq.Column('description', sq.String),
+            sq.Column('deleted', sq.Boolean),
+            sq.Column('status', sq.Integer),
+            extend_existing=True,
+        )
+
+    def create(self, data: domain.Customer) -> domain.Customer:
+        try:
+            new_customer = data.__dict__
+            new_customer.pop('id', None)
+            new_customer['deleted'] = False
+            new_customer['status'] = 0
+
+            query = self.customer.insert().values(new_customer).return_defaults()
+            customer = self.db.get_session().execute(query)
+            new_customer.update({"id": customer.inserted_primary_key[0]})
+            return self.__create_customer_dto(new_customer)
+
+        except sq.exc.SQLAlchemyError as err:
+            print(err)
+            return None
+
+    # def check_customer_exists(self, data: domain.Customer) -> bool:
+    #     customer_upper_name = str(data.name).strip().upper()
+    #     query = sql.select(self.customer).where(str(self.customer.c.name).upper() == customer_upper_name)
+    #     exists = self.db.get_session().execute(query).one_or_none()
+    #     return True if exists else False
+
+    def __create_customer_dto(self, customer: dict) -> domain.Customer:
+        customer = {key: customer.get(key) for key in self.customer_key}
+        return domain.Customer(**customer)
+
+    # def delete(self, customer_id: int) -> domain.Customer:
+    #     query = (
+    #         self.customer.update()
+    #         .where(self.customer.c.id == customer_id)
+    #         .values({"deleted": True})
+    #     )
+    #     self.db.get_session().execute(query)
+    #     query_deleted_customer = sq.sql.select(self.customer).where(self.customer.c.id == customer_id)
+    #     customer = self.db.get_session().execute(query_deleted_customer).one_or_none()
+    #     return self.__create_customer_dto(dict(customer)) if customer else None

V2/time_tracker/customers/interface.py

@@ -0,0 +1,2 @@
+# flake8: noqa
+from ._application import create_customer