feat(time-entries): Add endpoint to get summary of worked time

Author: Angeluz-07

commons/data_access_layer/cosmos_db.py

@@ -379,6 +379,11 @@ def __init__(self, status_code: int, description: str = None):
         self.description = description
 
 
+def init_app(app: Flask) -> None:
+    global cosmos_helper
+    cosmos_helper = CosmosDBFacade.from_flask_config(app)
+
+
 def current_datetime() -> datetime:
     return datetime.utcnow()
 
@@ -398,11 +403,6 @@ def generate_uuid4() -> str:
     return str(uuid.uuid4())
 
 
-def init_app(app: Flask) -> None:
-    global cosmos_helper
-    cosmos_helper = CosmosDBFacade.from_flask_config(app)
-
-
 def get_last_day_of_month(year: int, month: int) -> int:
     from calendar import monthrange
     return monthrange(year=year, month=month)[1]

time_tracker_api/time_entries/custom_modules/worked_time.py

@@ -0,0 +1,126 @@
+from datetime import datetime, timedelta
+from commons.data_access_layer.cosmos_db import (
+    current_datetime,
+    current_datetime_str,
+    datetime_str,
+    get_current_month,
+    get_current_year
+)
+
+
+def start_datetime_of_current_month() -> datetime:
+    return datetime(year=get_current_year(), month=get_current_month(), day=1,)
+
+
+def start_datetime_of_current_week() -> datetime:
+    today = current_datetime()
+    # WARNING : this gives monday datetime with hours at current hour
+    monday = today - timedelta(days=today.weekday())
+    return monday
+
+
+def start_datetime_of_current_day() -> datetime:
+    today = current_datetime()
+    today = today.replace(hour=0, minute=0, second=0, microsecond=000000)
+    return today
+
+
+def start_datetime_of_current_month_str() -> str:
+    return datetime_str(start_datetime_of_current_month())
+
+
+def str_to_datetime(
+    value: str, conversion_format: str = '%Y-%m-%dT%H:%M:%S.%f'
+) -> datetime:
+    return datetime.strptime(value, conversion_format)
+
+
+def date_range():
+    return {
+        "start_date": start_datetime_of_current_month_str(),
+        "end_date": current_datetime_str(),
+    }
+
+
+def filter_time_entries(
+    time_entries, start_date: datetime, end_date: datetime = current_datetime()
+):
+    return [
+        t
+        for t in time_entries
+        if start_date <= str_to_datetime(t.start_date) <= end_date
+        or start_date <= str_to_datetime(t.end_date) <= end_date
+    ]
+
+
+def stop_running_time_entry(time_entries):
+    for t in time_entries:
+        if t.end_date is None:
+            t.end_date = current_datetime_str()
+
+
+class WorkedTime:
+    def __init__(self, time_entries):
+        self.time_entries = time_entries
+
+    def total_time_in_seconds(self):
+        times = []
+
+        for t in self.time_entries:
+            start_datetime = str_to_datetime(t.start_date)
+            end_datetime = str_to_datetime(t.end_date)
+
+            elapsed_time = end_datetime - start_datetime
+            times.append(elapsed_time)
+
+        total_time = timedelta()
+        for time in times:
+            total_time += time
+
+        return total_time.total_seconds()
+
+    def hours(self):
+        return self.total_time_in_seconds() // 3600
+
+    def minutes(self):
+        return (self.total_time_in_seconds() % 3600) // 60
+
+    def seconds(self):
+        return (self.total_time_in_seconds() % 3600) % 60
+
+    def summary(self):
+        return {
+            "hours": self.hours(),
+            "minutes": self.minutes(),
+            "seconds": round(self.seconds(), 2),
+        }
+
+
+def worked_time_in_day(time_entries):
+    day_time_entries = filter_time_entries(
+        time_entries, start_date=start_datetime_of_current_day()
+    )
+    return WorkedTime(day_time_entries).summary()
+
+
+def worked_time_in_week(time_entries):
+    week_time_entries = filter_time_entries(
+        time_entries, start_date=start_datetime_of_current_week()
+    )
+    return WorkedTime(week_time_entries).summary()
+
+
+def worked_time_in_month(time_entries):
+    month_time_entries = filter_time_entries(
+        time_entries, start_date=start_datetime_of_current_month()
+    )
+    return WorkedTime(month_time_entries).summary()
+
+
+def summary(time_entries):
+    stop_running_time_entry(time_entries)
+    return {
+        'day': worked_time_in_day(time_entries),
+        'week': worked_time_in_week(time_entries),
+        'month': worked_time_in_month(time_entries),
+    }

time_tracker_api/time_entries/time_entries_model.py

@@ -5,12 +5,14 @@
 from azure.cosmos import PartitionKey
 from flask_restplus._http import HTTPStatus
 
+from .custom_modules import worked_time
+
 from commons.data_access_layer.cosmos_db import (
     CosmosDBDao,
     CosmosDBRepository,
     CustomError,
-    current_datetime_str,
     CosmosDBModel,
+    current_datetime_str,
     get_date_range_of_month,
     get_current_year,
     get_current_month,
@@ -105,7 +107,10 @@ def create_sql_date_range_filter(date_range: dict) -> str:
             return ''
 
     def find_all(
-        self, event_context: EventContext, conditions: dict, date_range: dict
+        self,
+        event_context: EventContext,
+        conditions: dict = {},
+        date_range: dict = {},
     ):
         custom_sql_conditions = []
         custom_sql_conditions.append(
@@ -332,6 +337,19 @@ def find_running(self):
             event_ctx.tenant_id, event_ctx.user_id
         )
 
+    def get_worked_time(self, conditions: dict = {}):
+        event_ctx = self.create_event_context(
+            "read", "Summary of worked time in the current month"
+        )
+        conditions.update({"owner_id": event_ctx.user_id})
+
+        time_entries = self.repository.find_all(
+            event_ctx,
+            conditions=conditions,
+            date_range=worked_time.date_range(),
+        )
+        return worked_time.summary(time_entries)
+
     @staticmethod
     def handle_date_filter_args(args: dict) -> dict:
         if 'month' and 'year' in args:

time_tracker_api/time_entries/time_entries_namespace.py

@@ -247,3 +247,16 @@ class ActiveTimeEntry(Resource):
     def get(self):
         """Find the time entry that is running"""
         return time_entries_dao.find_running()
+
+
+@ns.route('/summary')
+@ns.response(HTTPStatus.OK, 'Summary of worked time in the current month')
+@ns.response(
+    HTTPStatus.NOT_FOUND, 'There is no time entry in the current month'
+)
+class WorkedTimeSummary(Resource):
+    @ns.doc('summary_of_worked_time')
+    # @ns.marshal_with(time_entry)
+    def get(self):
+        """Find the summary of worked time"""
+        return time_entries_dao.get_worked_time()