Dev testing new

api/data_ingestion/api.py

@@ -16,10 +16,12 @@
     core,
     deletion_requests,
     email,
+    error_table,
     groups,
     qos,
     roles,
     schema,
+    school_registration,
     upload,
     users,
     utils,
@@ -136,10 +138,12 @@ 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)
 app.include_router(schema.router)
+app.include_router(school_registration.router)
 app.include_router(upload.router)
 app.include_router(users.router)
 app.include_router(utils.router)

api/data_ingestion/internal/school_registration.py

@@ -0,0 +1,111 @@
+import csv
+import io
+import json
+import logging
+
+import requests
+
+from data_ingestion.internal.storage import storage_client
+from data_ingestion.settings import settings
+from data_ingestion.utils.data_quality import get_metadata_path
+
+logger = logging.getLogger(__name__)
+
+
+GEOLOCATION_CSV_COLUMNS = [
+    "school_id_giga",
+    "school_id_govt",
+    "school_name",
+    "latitude",
+    "longitude",
+    "education_level",
+    "contact_name",
+    "contact_email",
+    "verification_status",
+]
+
+
+def write_registration_csv_to_adls(
+    payload: dict,
+    file_upload,
+    registration_metadata: dict,
+    mode: str = "create",
+) -> None:
+    """
+    Formats the school registration payload as a single-row CSV and uploads it to ADLS.
+    """
+    row = {
+        "school_id_giga": payload.get("giga_id_school", ""),
+        "school_id_govt": payload.get("school_id", ""),
+        "school_name": payload.get("school_name", ""),
+        "latitude": payload.get("latitude", ""),
+        "longitude": payload.get("longitude", ""),
+        "education_level": payload.get("education_level", ""),
+        "contact_name": payload.get("contact_name", ""),
+        "contact_email": payload.get("contact_email", ""),
+        "verification_status": registration_metadata.get(
+            "verification_status", "unverified"
+        ),
+    }
+
+    output = io.StringIO()
+    writer = csv.DictWriter(output, fieldnames=GEOLOCATION_CSV_COLUMNS)
+    writer.writeheader()
+    writer.writerow(row)
+    csv_bytes = output.getvalue().encode("utf-8")
+
+    blob_client = storage_client.get_blob_client(file_upload.upload_path)
+    blob_client.upload_blob(
+        csv_bytes,
+        overwrite=True,
+        metadata={
+            "country": file_upload.country,
+            "uploader_email": file_upload.uploader_email,
+            "school_id_giga": payload.get("giga_id_school", ""),
+            "mode": mode,
+        },
+    )
+
+    metadata_payload = {
+        "country": file_upload.country,
+        "uploader_email": file_upload.uploader_email,
+        "dataset": "geolocation",
+        "mode": mode,
+        **registration_metadata,
+    }
+    metadata_path = get_metadata_path(file_upload.upload_path)
+    metadata_blob_client = storage_client.get_blob_client(metadata_path)
+    metadata_blob_client.upload_blob(
+        json.dumps(metadata_payload, indent=2).encode(),
+        overwrite=True,
+    )
+
+
+def call_meter_soft_delete(school_id_giga: str, rejection_reason: str = None) -> None:
+    """
+    Calls the GigaMeter API to soft-delete a school_new_registration record.
+    """
+    if not settings.GIGAMETER_API_BASE_URL or not settings.GIGAMETER_API_TOKEN:
+        logger.warning(
+            "GIGAMETER_API_BASE_URL or GIGAMETER_API_TOKEN not configured. "
+            "Skipping soft-delete callback for school_id_giga=%s",
+            school_id_giga,
+        )
+        return
+
+    url = f"{settings.GIGAMETER_API_BASE_URL}/api/v1/school-registrations/rejection"
+    headers = {
+        "Authorization": f"Bearer {settings.GIGAMETER_API_TOKEN}",
+        "Content-Type": "application/json",
+    }
+    payload = {
+        "giga_id_school": school_id_giga,
+        "is_deleted": True,
+        "rejection_reason": rejection_reason or "Rejected by admin",
+    }
+
+    response = requests.put(url, headers=headers, json=payload, timeout=10)
+    logger.info(
+        "GigaMeter soft-delete successful for school_id_giga=%s", school_id_giga
+    )
+    return response.json()

api/data_ingestion/migrations/versions/2025_11_27_1050_49a684bb43f3_.py

@@ -0,0 +1,33 @@
+"""
+
+Revision ID: 49a684bb43f3
+Revises: a3c6ab14b3f8
+Create Date: 2025-11-27 10:50:01.201361
+
+"""
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = '49a684bb43f3'
+down_revision: str | None = 'a3c6ab14b3f8'
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('approval_requests', sa.Column('upload_id', sa.String(), nullable=True))
+    op.create_index(op.f('ix_approval_requests_upload_id'), 'approval_requests', ['upload_id'], unique=False)
+    op.create_foreign_key('fk_approval_requests_upload_id_file_uploads', 'approval_requests', 'file_uploads', ['upload_id'], ['id'])
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint('fk_approval_requests_upload_id_file_uploads', 'approval_requests', type_='foreignkey')
+    op.drop_index(op.f('ix_approval_requests_upload_id'), table_name='approval_requests')
+    op.drop_column('approval_requests', 'upload_id')
+    # ### end Alembic commands ###

api/data_ingestion/migrations/versions/2026_03_30_1201_7b89fb20822f_.py

@@ -0,0 +1,109 @@
+"""
+
+Revision ID: 7b89fb20822f
+Revises: 49a684bb43f3
+Create Date: 2026-03-30 12:01:25.183717
+
+"""
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = '7b89fb20822f'
+down_revision: str | None = '49a684bb43f3'
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.execute("DO $$ BEGIN IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'dq_mode') THEN CREATE TYPE dq_mode AS ENUM ('uploaded', 'master'); END IF; END $$;")
+    op.execute("DO $$ BEGIN IF NOT EXISTS (SELECT 1 FROM pg_type WHERE typname = 'dqmodeenum') THEN CREATE TYPE dqmodeenum AS ENUM ('uploaded', 'master'); END IF; END $$;")
+
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    tables = inspector.get_table_names()
+    if 'dq_runs' not in tables:
+        op.create_table('dq_runs',
+            sa.Column('id', sa.Integer(), nullable=False),
+            sa.Column('upload_id', sa.String(), nullable=False),
+            sa.Column('dq_mode', postgresql.ENUM('uploaded', 'master', name='dq_mode', create_type=False), nullable=False, server_default='uploaded'),
+            sa.Column('status', sa.String(), nullable=False),
+            sa.Column('dagster_run_id', sa.String(), nullable=True),
+            sa.Column('result_path', sa.String(), nullable=True),
+            sa.ForeignKeyConstraint(['upload_id'], ['file_uploads.id'], ondelete='CASCADE'),
+            sa.PrimaryKeyConstraint('id')
+        )
+        op.create_index(op.f('ix_dq_runs_upload_id'), 'dq_runs', ['upload_id'], unique=False)
+
+    columns_audit = [c['name'] for c in inspector.get_columns('approval_request_audit_log')]
+    if 'dq_mode' not in columns_audit:
+        op.add_column('approval_request_audit_log',
+            sa.Column('dq_mode', postgresql.ENUM('uploaded', 'master', name='dqmodeenum', create_type=False), nullable=False, server_default='uploaded')
+        )
+
+    columns_req = [c['name'] for c in inspector.get_columns('approval_requests')]
+    if 'dq_mode' not in columns_req:
+        op.add_column('approval_requests',
+            sa.Column('dq_mode', postgresql.ENUM('uploaded', 'master', name='dqmodeenum', create_type=False), nullable=False, server_default='uploaded')
+        )
+
+    op.drop_constraint(op.f('uq_country_dataset'), 'approval_requests', type_='unique')
+    op.create_unique_constraint('uq_country_dataset', 'approval_requests', ['country', 'dataset', 'upload_id'])
+    op.alter_column('file_uploads', 'dq_full_path',
+               existing_type=sa.TEXT(),
+               type_=sa.String(),
+               existing_nullable=True)
+    op.alter_column('file_uploads', 'bronze_path',
+               existing_type=sa.TEXT(),
+               type_=sa.String(),
+               existing_nullable=True)
+    op.alter_column('file_uploads', 'is_processed_in_staging',
+               existing_type=sa.BOOLEAN(),
+               nullable=False,
+               existing_server_default=sa.text('false'))
+    op.alter_column('qos_school_connectivity', 'send_date_in',
+               existing_type=postgresql.ENUM('BODY', 'QUERY_PARAMETERS', 'NONE', name='senddateinenum'),
+               nullable=False,
+               existing_server_default=sa.text("'NONE'::senddateinenum"))
+    op.alter_column('qos_school_connectivity', 'school_id_giga_govt_key',
+               existing_type=sa.TEXT(),
+               type_=sa.String(),
+               existing_nullable=False,
+               existing_server_default=sa.text("'school_id_giga'::text"))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.alter_column('qos_school_connectivity', 'school_id_giga_govt_key',
+               existing_type=sa.String(),
+               type_=sa.TEXT(),
+               existing_nullable=False,
+               existing_server_default=sa.text("'school_id_giga'::text"))
+    op.alter_column('qos_school_connectivity', 'send_date_in',
+               existing_type=postgresql.ENUM('BODY', 'QUERY_PARAMETERS', 'NONE', name='senddateinenum'),
+               nullable=True,
+               existing_server_default=sa.text("'NONE'::senddateinenum"))
+    op.alter_column('file_uploads', 'is_processed_in_staging',
+               existing_type=sa.BOOLEAN(),
+               nullable=True,
+               existing_server_default=sa.text('false'))
+    op.alter_column('file_uploads', 'bronze_path',
+               existing_type=sa.String(),
+               type_=sa.TEXT(),
+               existing_nullable=True)
+    op.alter_column('file_uploads', 'dq_full_path',
+               existing_type=sa.String(),
+               type_=sa.TEXT(),
+               existing_nullable=True)
+    op.drop_constraint('uq_country_dataset', 'approval_requests', type_='unique')
+    op.create_unique_constraint(op.f('uq_country_dataset'), 'approval_requests', ['country', 'dataset'], postgresql_nulls_not_distinct=False)
+    op.drop_column('approval_requests', 'dq_mode')
+    op.drop_column('approval_request_audit_log', 'dq_mode')
+    op.drop_index(op.f('ix_dq_runs_upload_id'), table_name='dq_runs')
+    op.drop_table('dq_runs')
+    # ### end Alembic commands ###

api/data_ingestion/models/__init__.py

@@ -1,6 +1,7 @@
 from .approval_requests import ApprovalRequest
 from .base import BaseModel
 from .deletion_requests import DeletionRequest
+from .dq_run import DQRun
 from .file_upload import FileUpload
 from .ingest_api_qos import ApiConfiguration, SchoolConnectivity, SchoolList
 from .users import Role, User, UserRoleAssociation
@@ -16,4 +17,5 @@
     "User",
     "Role",
     "UserRoleAssociation",
+    "DQRun",
 ]

api/data_ingestion/models/approval_requests.py

@@ -1,7 +1,13 @@
 from datetime import datetime
 
 from pydantic import UUID4
-from sqlalchemy import VARCHAR, DateTime, ForeignKey, UniqueConstraint, func
+from sqlalchemy import (
+    VARCHAR,
+    DateTime,
+    ForeignKey,
+    UniqueConstraint,
+    func,
+)
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from .base import BaseModel

api/data_ingestion/models/dq_run.py

@@ -0,0 +1,36 @@
+from enum import Enum
+
+from sqlalchemy import ForeignKey
+from sqlalchemy.dialects.postgresql import ENUM as SQLEnum
+from sqlalchemy.orm import Mapped, mapped_column
+
+from .base import BaseModel
+
+
+class DQModeEnum(str, Enum):
+    uploaded = "uploaded"
+    master = "master"
+
+
+class DQRun(BaseModel):
+    __tablename__ = "dq_runs"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+
+    upload_id: Mapped[str] = mapped_column(
+        ForeignKey("file_uploads.id", ondelete="CASCADE"),
+        index=True,
+        nullable=False,
+    )
+
+    dq_mode: Mapped[DQModeEnum] = mapped_column(
+        SQLEnum(DQModeEnum, name="dq_mode"),
+        nullable=False,
+        default=DQModeEnum.uploaded,
+    )
+
+    status: Mapped[str] = mapped_column(nullable=False)
+
+    dagster_run_id: Mapped[str | None] = mapped_column(nullable=True)
+
+    result_path: Mapped[str | None] = mapped_column(nullable=True)

api/data_ingestion/routers/approval_requests.py

@@ -12,6 +12,7 @@
 from data_ingestion.db.primary import get_db as get_primary_db
 from data_ingestion.db.trino import get_db
 from data_ingestion.internal.auth import azure_scheme
+from data_ingestion.internal.school_registration import call_meter_soft_delete
 from data_ingestion.internal.storage import storage_client
 from data_ingestion.models import (
     ApprovalRequest,
@@ -27,6 +28,7 @@
     UploadListing,
 )
 from data_ingestion.schemas.core import PagedResponseSchema
+from data_ingestion.utils.nocodb import update_nocodb_record_by_field
 
 router = APIRouter(
     prefix="/api/approval-requests",
@@ -480,6 +482,28 @@ async def submit_upload_review(
         body.rejected_rows, staging, upload_id, db
     )
 
+    # Call GigaMeter soft-delete and update NocoDB for rejected GigaMeter registrations
+    if rejected_change_ids and file_upload.source == "gigameter":
+        # Extract school_id_giga from rejected change_ids
+        for change_id in rejected_change_ids:
+            school_id_giga = change_id.split("|")[0]
+            # Update NocoDB status to rejected
+            update_nocodb_record_by_field(
+                table_name="SchoolRegistrations",
+                field_name="giga_id_school",
+                field_value=school_id_giga,
+                update_data={
+                    "verification_status": "rejected",
+                    "rejected_on": str(datetime.now(UTC)),
+                    "rejection_reason": "Rejected by admin during manual review",
+                },
+            )
+            # Call GigaMeter soft-delete
+            call_meter_soft_delete(
+                school_id_giga=school_id_giga,
+                rejection_reason="Rejected by admin during manual review",
+            )
+
     # Create the audit log first so its ID can be included in the approval payload.
     approval_request_log_id = None
     if approval_request: