feat: TT-357 Change Database Connection and Update, Create Funtions

Author: ararcos

V2/time_entries/_infrastructure/_data_persistence/_activities_sql_dao.py

@@ -1,49 +1,49 @@
-from sqlalchemy.engine.cursor import LegacyCursorResult
 from time_entries._domain import ActivitiesDao, Activity
-from sqlalchemy import create_engine,Table, Column, String, MetaData
 from sqlalchemy.sql import select
+from ..database import db
 import dataclasses
-import json
 import typing
 
-
 class ActivitiesSQLDao(ActivitiesDao):
     def __init__(self):
         self.activity_keys = [
             field.name for field in dataclasses.fields(Activity)
         ]
-        self.engine = create_engine('postgresql://postgres:root@localhost/time-tracker')
-        self.conn = self.engine.connect()
-        metadata_obj = MetaData()
-        self.activity = Table('activity', metadata_obj,
-            Column('id', String, primary_key=True),
-            Column('name', String),
-            Column('description', String),
-            Column('deleted', String),
-            Column('status', String),
-            Column('tenant_id', String),
-        )
+        db.db_init()
 
 
-
+   #GET Activity by id
     def get_by_id(self, activity_id: str) -> Activity:
-        query = select(self.activity).where(self.activity.c.id == activity_id)
-        activity = self.conn.execute(query).one()
+        """search for the activity by the given id
+        parameters: str (the id string of the searched activity)
+        returns: Activity (The activity model with the response if found or a void if not)."""
+        query = select(db.activity).where(db.activity.c.id == activity_id)
+        activity = db.connection.execute(query).one()
         return self.__create_activity_dto(dict(activity)) if activity else None
 
     def get_all(self) -> typing.List[Activity]:
-        query = select(self.activity)
-        result = self.conn.execute(query)
+        query = select(db.activity)
+        result = db.connection.execute(query)
         return [
             self.__create_activity_dto(dict(activity))
             for activity in result
         ]
 
+    def create_activity(self, activity_data: dict) -> Activity:
+        query = db.activity.insert().values(activity_data)
+        activity = db.connection.execute(query)
+        return self.__create_activity_dto(activity.last_inserted_params())
+    
     def delete(self, activity_id: str) -> Activity:
-        query = self.activity.update().where(self.activity.c.id == activity_id).values(status = 'inactive')
-        self.conn.execute(query)
+        query = db.activity.update().where(db.activity.c.id == activity_id).values(status = 'inactive')
+        db.connection.execute(query)
         return self.get_by_id(activity_id);
 
+    def update(self, activity_id: str, new_activity: dict) -> Activity:
+        query = db.activity.update().where(db.activity.c.id==activity_id).values(new_activity)
+        db.connection.execute(query)
+        return self.get_by_id(activity_id)
+
     def __create_activity_dto(self, activity: dict) -> Activity:
-        activity = {key: activity.get(key) for key in self.activity_keys}
+        activity = {key: activity.get(key).hex if key == "id" else activity.get(key) for key in self.activity_keys}
         return Activity(**activity)

V2/time_entries/_infrastructure/database/__init__.py

@@ -0,0 +1 @@
+from .data_access_layer_sql import db

V2/time_entries/_infrastructure/database/data_access_layer_sql.py

@@ -0,0 +1,59 @@
+from sqlalchemy.types import TypeDecorator, CHAR
+from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy import create_engine,Table, Column, String, MetaData
+
+import uuid
+class GUID(TypeDecorator):
+    """Platform-independent GUID type.
+    Uses PostgreSQL's UUID type, otherwise uses
+    CHAR(32), storing as stringified hex values.
+    """
+    impl = CHAR
+
+    def load_dialect_impl(self, dialect):
+        if dialect.name == 'postgresql':
+            return dialect.type_descriptor(UUID())
+        else:
+            return dialect.type_descriptor(CHAR(32))
+
+    def process_bind_param(self, value, dialect):
+        if value is None:
+            return value
+        elif dialect.name == 'postgresql':
+            return str(value)
+        else:
+            if not isinstance(value, uuid.UUID):
+                return "%.32x" % uuid.UUID(value).int
+            else:
+                # hexstring
+                return "%.32x" % value.int
+
+    def process_result_value(self, value, dialect):
+        if value is None:
+            return value
+        else:
+            if not isinstance(value, uuid.UUID):
+                value = uuid.UUID(value)
+            return value
+
+class DataAccessLayerSQL:
+    connection = None
+    engine = None
+    conn_string = 'postgresql://postgres:root@localhost/time-tracker'
+    metadata = MetaData()
+    activity = Table('activity', metadata,
+                    Column('id',GUID() , primary_key=True,default= uuid.uuid4),
+                    Column('name', String),
+                    Column('description', String),
+                    Column('deleted', String),
+                    Column('status', String),
+                    Column('tenant_id', String),
+              )
+
+    def db_init(self, conn_string: str = conn_string): 
+        self.engine = create_engine(conn_string)
+        self.metadata.create_all(self.engine)
+        self.connection = self.engine.connect()
+
+db = DataAccessLayerSQL() 
+