DataOps-platform/app/models/metadata_review.py

from __future__ import annotations

from datetime import datetime
from typing import Any, Optional

from sqlalchemy.dialects.postgresql import JSONB

from app import db


class MetadataReviewRecord(db.Model):
    __tablename__ = "metadata_review_records"
    __table_args__ = {"schema": "public"}

    id = db.Column(db.BigInteger, primary_key=True)
    record_type = db.Column(db.String(20), nullable=False)  # redundancy | change
    source = db.Column(db.String(50), nullable=False, default="ddl")
    business_domain_id = db.Column(db.BigInteger, nullable=True)

    new_meta = db.Column(JSONB, nullable=False)
    candidates = db.Column(JSONB, nullable=False, default=list)
    old_meta = db.Column(JSONB, nullable=True)

    status = db.Column(db.String(20), nullable=False, default="pending")
    resolution_action = db.Column(db.String(30), nullable=True)
    resolution_payload = db.Column(JSONB, nullable=True)

    notes = db.Column(db.Text, nullable=True)
    created_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
    resolved_at = db.Column(db.DateTime, nullable=True)
    resolved_by = db.Column(db.String(100), nullable=True)

    def to_dict(self) -> dict[str, Any]:
        return {
            "id": self.id,
            "record_type": self.record_type,
            "source": self.source,
            "business_domain_id": self.business_domain_id,
            "new_meta": self.new_meta,
            "candidates": self.candidates,
            "old_meta": self.old_meta,
            "status": self.status,
            "resolution_action": self.resolution_action,
            "resolution_payload": self.resolution_payload,
            "notes": self.notes,
            "created_at": self.created_at.isoformat() if self.created_at else None,
            "updated_at": self.updated_at.isoformat() if self.updated_at else None,
            "resolved_at": self.resolved_at.isoformat() if self.resolved_at else None,
            "resolved_by": self.resolved_by,
        }


class MetadataVersionHistory(db.Model):
    __tablename__ = "metadata_version_history"
    __table_args__ = {"schema": "public"}

    id = db.Column(db.BigInteger, primary_key=True)
    meta_id = db.Column(db.BigInteger, nullable=False)
    change_source = db.Column(db.String(50), nullable=False, default="ddl")

    before_snapshot = db.Column(JSONB, nullable=False)
    after_snapshot = db.Column(JSONB, nullable=False)

    created_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
    created_by = db.Column(db.String(100), nullable=True)

    def to_dict(self) -> dict[str, Any]:
        return {
            "id": self.id,
            "meta_id": self.meta_id,
            "change_source": self.change_source,
            "before_snapshot": self.before_snapshot,
            "after_snapshot": self.after_snapshot,
            "created_at": self.created_at.isoformat() if self.created_at else None,
            "created_by": self.created_by,
        }


def update_review_record_resolution(
    record: MetadataReviewRecord,
    action: str,
    payload: Optional[dict[str, Any]] = None,
    resolved_by: Optional[str] = None,
    notes: Optional[str] = None,
) -> None:
    record.status = "resolved" if action != "ignore" else "ignored"
    record.resolution_action = action
    record.resolution_payload = payload or {}
    record.resolved_by = resolved_by
    record.resolved_at = datetime.utcnow()
    record.updated_at = datetime.utcnow()
    if notes is not None:
        record.notes = notes