Merge branch 'master' of https://github.com/palladiumkenya/datamap-ba…

…ckend into feat/transmission

models/models.py

@@ -13,6 +13,7 @@ class AccessCredentials(Model):
     id = columns.UUID(primary_key=True, default=uuid.uuid1)
     conn_string = columns.Text(required=True)
     name = columns.Text(required=True)
+    system_id = columns.UUID(required=True, index=True)
 
     is_active = columns.Boolean(required=True, default=True)
     created_at = columns.DateTime(required=True, default=datetime.utcnow())
@@ -66,6 +67,7 @@ def save(self):
         self.updated_at = datetime.utcnow()
         super().save()
 
+
 class IndicatorHistory(Model):
     __keyspace__ = 'datamap'
     __table_name__ = 'indicator_history'
@@ -83,6 +85,7 @@ def save(self):
         self.created_at = datetime.utcnow()
         super().save()
 
+
 class TransmissionHistory(Model):
     __keyspace__ = 'datamap'
     __table_name__ = 'transmission_history'

routes/access_api.py

@@ -5,8 +5,9 @@
 from sqlalchemy.exc import OperationalError
 
 from database import database
-from models.models import AccessCredentials
-from serializers.access_credentials_serializer import access_credential_list_entity
+from models.models import AccessCredentials, SiteConfig
+from serializers.access_credentials_serializer import access_credential_list_entity, systems_list_entity, system_entity, \
+    access_credential_entity
 
 router = APIRouter()
 
@@ -20,23 +21,36 @@ async def available_connections():
 
 @router.get('/active_connection')
 async def active_connection():
-    return (
-        AccessCredentials.objects()
+    active_credentials = (
+        AccessCredentials
+        .objects()
         .filter(is_active=True)
         .allow_filtering()
         .first()
     )
 
+    if active_credentials is None:
+        return {"error": "No active credentials found"}
+
+    credentials = access_credential_entity(active_credentials)
+    if credentials is not None:
+        system = SiteConfig.objects().filter(id=credentials['system_id']).first()
+        credentials['system'] = system_entity(system)
+        return credentials
+
+    return {"error": "Failed to process active credentials"}
+
 
 class SaveDBConnection(BaseModel):
     conn_string: str = Field(..., description="Type of the database (e.g., 'mysql', 'postgresql')")
-    name: str = Field(..., description="Database username")
+    name: str = Field(..., description="Connection name")
+    system_id: str = Field(..., description="System ID")
 
 
 @router.post('/add_connection')
 async def add_connection(data: SaveDBConnection):
     try:
-        AccessCredentials.create(conn_string=data.conn_string, name=data.name)
+        AccessCredentials.create(conn_string=data.conn_string, name=data.name, system_id=data.system_id)
         return {'success': True, 'message': 'Connection added successfully'}
     except Exception as e:
         raise HTTPException(status_code=500, detail=e) from e
@@ -90,3 +104,13 @@ async def test_db_connection(data: DBConnectionRequest):
         return {"status": "Database connection successful"}
     else:
         raise HTTPException(status_code=500, detail=error_message)
+
+
+@router.get("/dictionary/systems")
+def get_system_list():
+    systems = SiteConfig.objects.filter(is_active=True).allow_filtering().all()
+    systems_list = systems_list_entity(systems)
+    return {
+        "data": systems_list,
+        "success": True
+    }

routes/dictionary_mapper_api.py

@@ -1,3 +1,5 @@
+import re
+
 from fastapi import  Depends, HTTPException
 from sqlalchemy import create_engine, inspect, MetaData, Table,text
 from sqlalchemy.exc import SQLAlchemyError
@@ -349,7 +351,7 @@ async def load_data(baselookup:str, db_session: Session = Depends(get_db)):
         with db_session as session:
             result = session.execute(query)
 
-            columns=result.keys()
+            columns = result.keys()
             baseRepoLoaded = [dict(zip(columns,row)) for row in result]
 
             # processed_results = [convert_datetime_to_iso(convert_none_to_null(result)) for result in baseRepoLoaded]
@@ -358,7 +360,6 @@ async def load_data(baselookup:str, db_session: Session = Depends(get_db)):
             batch = BatchStatement()
             cass_session.execute("TRUNCATE TABLE %s;" %(baselookup))
             for data in processed_results:
-
                 quoted_values = [
                     'NULL' if value is None
                     else f"'{value}'" if isinstance(value, str)
@@ -368,7 +369,7 @@ async def load_data(baselookup:str, db_session: Session = Depends(get_db)):
                     for key, value in data.items()
                 ]
 
-                idColumn = baselookup +"_id"
+                idColumn = baselookup + "_id"
 
                 query = f"""
                            INSERT INTO {baselookup} ({idColumn}, {", ".join(tuple(data.keys()))})
@@ -390,9 +391,32 @@ async def load_data(baselookup:str, db_session: Session = Depends(get_db)):
             # loadedHistory.save()
             # TransmissionHistory.objects(id=loadedHistory.id).update(ended_at=datetime.utcnow())
 
-            return {"data":baseRepoLoaded}
+            return {"data": [expected_variables_dqa(data, baselookup) for data in baseRepoLoaded]}
     except Exception as e:
         log.error("Error loading data ==> %s", str(e))
         raise HTTPException(status_code=500, detail="Error loading data:" + str(e))
 
 
+def expected_variables_dqa(data, lookup):
+    valid_match = True
+    try:
+        if dictionary := DataDictionaries.objects.filter(name=lookup).first():
+            dictionary_terms = DataDictionaryTerms.objects.filter(dictionary_id=dictionary.id).all()
+            for term in dictionary_terms:
+                column_data = data.get(term.term)
+                if is_valid_regex(term.expected_values) and not re.match(pattern=term.expected_values, string=str(column_data)):
+                    valid_match = False
+
+    except Exception as e:
+        log.error(f"Error {str(e)}")
+
+    data['valid_match'] = valid_match
+    return data
+
+
+def is_valid_regex(pattern):
+    try:
+        re.compile(pattern)
+        return True
+    except re.error:
+        return False

serializers/access_credentials_serializer.py

@@ -4,10 +4,24 @@ def access_credential_entity(credential) -> dict:
         "conn_string": str(credential["conn_string"]),
         "name": str(credential["name"]),
         "is_active": bool(credential["is_active"]),
+        "system_id": str(credential["system_id"]),
         "created_at": credential["created_at"],
         "updated_at": credential["updated_at"]
     }
 
 
+def system_entity(system) -> dict:
+    return {
+        "id": str(system["id"]),
+        "name": str(system["site_name"]),
+        "site_id": str(system["site_id"]),
+        "primary_system": str(system["primary_system"])
+    }
+
+
 def access_credential_list_entity(credentials) -> list:
     return [access_credential_entity(credential) for credential in credentials]
+
+
+def systems_list_entity(systems) -> list:
+    return [system_entity(system) for system in systems]