feat: TT-365 create function serverless

Author: dsgarcia8

V2/serverless.yml

@@ -46,3 +46,13 @@ functions:
             - GET
           route: activities/{id:?}
           authLevel: anonymous
+
+  create_activity:
+    handler: time_entries/interface.create_activity
+    events:
+      - http: true
+        x-azure-settings:
+          methods:
+            - POST
+          route: activities/create
+          authLevel: anonymous

V2/time_entries/_application/__init__.py

@@ -1 +1 @@
-from ._activities import get_activities
+from ._activities import get_activities, create_activity

V2/time_entries/_application/_activities/__init__.py

@@ -1 +1,2 @@
 from ._get_activities import get_activities
+from ._create_activity import create_activity

V2/time_entries/_application/_activities/_create_activity.py

@@ -0,0 +1,48 @@
+from time_entries._infrastructure import ActivitiesJsonDao
+from time_entries._domain import ActivityService, _use_cases
+
+import azure.functions as func
+import json
+import logging
+
+JSON_PATH = (
+    'V2/time_entries/_infrastructure/_data_persistence/activities_data.json'
+)
+
+new_activity =     {
+        'id': 'c61a4a49-3364-49a3-a666-0c5f2d15072b',
+        'name': 'Activity Demo create',
+        'tenant_id': 'cc925a5d-9644-4a4f-8d99-0bee49aadd05',
+        'description': 'test demo create an new activity',
+        'status': 'active',
+        'deleted': 'b4327ba6-9f96-49ee-a9ac-3c1edf525172',
+    }
+
+def create_activity(req: func.HttpRequest) -> func.HttpResponse:
+    logging.info(
+        'Python HTTP trigger function processed a request to create an activity.'
+    )
+    #activity_id = req.route_params.get('id')
+    status_code = 200
+
+    if status_code == 200:
+        response = _create_activity(new_activity)
+        if response == b'Not Found':
+            status_code = 404
+    else:
+        response = b'Not Found'
+
+    return func.HttpResponse(
+        body=response, status_code=status_code, mimetype="application/json"
+    )
+
+def _create_activity(activity_data: dict) -> str:
+    activity_use_case = _use_cases.CreateActivityUseCase(
+        _create_activity_service(JSON_PATH)
+    )
+    activity = activity_use_case.create_activity(activity_data)
+    return json.dumps(activity.__dict__) if activity else b'Not Found'
+
+def _create_activity_service(path: str):
+    activity_json = ActivitiesJsonDao(path)
+    return ActivityService(activity_json)

V2/time_entries/_domain/_persistence_contracts/_activities_dao.py

@@ -11,3 +11,7 @@ def get_by_id(self, id: str) -> Activity:
     @abc.abstractmethod
     def get_all(self) -> typing.List[Activity]:
         pass
+
+    @abc.abstractmethod
+    def create_activity(self, activity_data: dict) -> Activity:
+        pass

V2/time_entries/_domain/_services/_activity.py

@@ -11,3 +11,6 @@ def get_by_id(self, activity_id: str) -> Activity:
 
     def get_all(self) -> typing.List[Activity]:
         return self.activities_dao.get_all()
+
+    def create_activity(self, activity_data: dict) -> Activity:
+        return self.activities_dao.create_activity(activity_data)

V2/time_entries/_domain/_use_cases/__init__.py

@@ -1,2 +1,3 @@
 from ._get_activities_use_case import GetActivitiesUseCase
 from ._get_activity_by_id_use_case import GetActivityUseCase
+from ._create_activity_use_case import CreateActivityUseCase

V2/time_entries/_domain/_use_cases/_create_activity_use_case.py

@@ -0,0 +1,11 @@
+from time_entries._domain import ActivityService, Activity
+import typing
+
+
+class CreateActivityUseCase:
+    def __init__(self, activity_service: ActivityService):
+        self.activity_service = activity_service
+
+    def create_activity(self, activity_data: dict ) -> Activity:
+        return self.activity_service.create_activity(activity_data)
+

V2/time_entries/_infrastructure/_data_persistence/_activities_json_dao.py

@@ -25,6 +25,12 @@ def get_all(self) -> typing.List[Activity]:
             for activity in self.__get_activities_from_file()
         ]
 
+    def create_activity(self, newactivity: Activity) -> typing.List[Activity]:
+        return [
+            self.__create_activity_dto(activity)
+            for activity in self.__get_activities_from_file()
+        ]
+
     def __get_activities_from_file(self) -> typing.List[dict]:
         try:
             file = open(self.json_data_file_path)

V2/time_entries/interface.py

@@ -1 +1,2 @@
-from ._application import get_activities
+from ._application import get_activities, create_activity
+