feat: Error table API

api/data_ingestion/api.py

@@ -16,6 +16,7 @@
     core,
     deletion_requests,
     email,
+    error_table,
     groups,
     qos,
     roles,
@@ -136,6 +137,7 @@ async def _ensure_local_dev_user():
 app.include_router(core.router)
 app.include_router(deletion_requests.router)
 app.include_router(email.router)
+app.include_router(error_table.router)
 app.include_router(groups.router)
 app.include_router(qos.router)
 app.include_router(roles.router)

api/data_ingestion/internal/schema.py

@@ -1,3 +1,5 @@
+import uuid
+
 from fastapi import BackgroundTasks
 from fastapi.encoders import jsonable_encoder
 from loguru import logger
@@ -30,7 +32,12 @@ async def get_schemas(
     )
     mappings = res.mappings().all()
     schemas = [
-        *[m["table_name"] for m in mappings],
+        *(
+            "school_geolocation"
+            if m["table_name"] == "school_geolocation_metadata"
+            else m["table_name"]
+            for m in mappings
+        ),
         "school_geolocation_qos",
         "school_geolocation_update",
     ]
@@ -41,6 +48,88 @@ async def get_schemas(
     return schemas
 
 
+def _should_skip_column(name: str, column_name: str) -> bool:
+    """Determine if a column should be skipped based on table context or system rules."""
+    if "geolocation" in name and column_name == "school_id_giga":
+        return True
+
+    # System generated columns filter
+    if column_name in [
+        "giga_sync_id",
+        "country",
+        "country_code",
+        "created_at",
+        "file_size_bytes",
+        "giga_sync_uploaded_at",
+        "raw_file_path",
+        "schema_name",
+    ]:
+        return True
+    return False
+
+
+def _apply_schema_overrides(name: str, col: SchemaColumn):
+    """Apply business-specific metadata overrides to a schema column."""
+    # Important fields tagging
+    if col.name in [
+        "school_id_govt",
+        "school_name",
+        "education_level_govt",
+        "latitude",
+        "longitude",
+    ]:
+        col.is_important = True
+    elif col.is_important is None:
+        col.is_important = False
+
+    if col.primary_key is None:
+        col.primary_key = False
+
+    # Nullability overrides for specific operational views
+    if name == "school_geolocation_qos" and col.name == "education_level_govt":
+        col.is_nullable = True
+
+    if name == "school_geolocation_update" and col.name != "school_id_govt":
+        col.is_nullable = True
+
+
+def _inject_missing_core_fields(name: str, schema: list[SchemaColumn]):
+    """Ensure core geolocation metadata fields exist in the schema."""
+    if not name.startswith("school_geolocation"):
+        return
+
+    existing_names = {s.name for s in schema}
+    core_fields = [
+        (
+            "school_id_govt",
+            "varchar",
+            False,
+            True,
+            True,
+            "Government unique identifier",
+        ),
+        ("school_name", "varchar", False, True, False, "Official school name"),
+        ("education_level_govt", "varchar", False, True, False, "Education level"),
+        ("latitude", "double", False, True, False, "Latitude"),
+        ("longitude", "double", False, True, False, "Longitude"),
+    ]
+    for name_f, dtype, nullable, important, pk, desc in core_fields:
+        if name_f not in existing_names:
+            schema.append(
+                SchemaColumn(
+                    id=str(uuid.uuid4()),
+                    name=name_f,
+                    data_type=dtype,
+                    is_nullable=nullable,
+                    is_important=important,
+                    is_system_generated=False,
+                    primary_key=pk,
+                    description=desc,
+                    license="ODBL",
+                )
+            )
+
+
 def get_schema(
     name: str,
     db: Session,
@@ -49,7 +138,7 @@ def get_schema(
     table_name = name
 
     if name.startswith("school_geolocation"):
-        table_name = "school_geolocation"
+        table_name = "school_geolocation_metadata"
 
     res = db.execute(
         select("*")
@@ -72,29 +161,14 @@ def get_schema(
         schema_column = SchemaColumn(**mapping)
         logger.info(schema_column.model_dump())
 
-        if "geolocation" in name and schema_column.name == "school_id_giga":
+        if _should_skip_column(name, schema_column.name):
             continue
 
-        if schema_column.is_important is None:
-            schema_column.is_important = False
-
-        if schema_column.primary_key is None:
-            schema_column.primary_key = False
-
-        if (
-            name == "school_geolocation_qos"
-            and schema_column.name == "education_level_govt"
-        ):
-            schema_column.is_nullable = True
-
-        if (
-            name == "school_geolocation_update"
-            and schema_column.name != "school_id_govt"
-        ):
-            schema_column.is_nullable = True
-
+        _apply_schema_overrides(name, schema_column)
         schema.append(schema_column)
 
+    _inject_missing_core_fields(name, schema)
+
     schema = sorted(schema, key=sort_schema_columns_key)
 
     if background_tasks is not None: