DataOps-platform/deployment/app/models/semantic_governance.py

"""SQLAlchemy mappings for generic semantic governance."""

from sqlalchemy.dialects.postgresql import JSONB, UUID

from app import db
from app.core.common.identifiers import new_governance_uid


class SemanticAsset(db.Model):
    __tablename__ = "semantic_assets"
    __table_args__ = (
        db.UniqueConstraint("asset_kind", "code"),
        {"schema": "public"},
    )

    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
    asset_kind = db.Column(db.String(30), nullable=False)
    code = db.Column(db.String(120), nullable=False)
    name = db.Column(db.String(300), nullable=False)
    owner_uid = db.Column(UUID(as_uuid=False), nullable=False)
    business_domain_uid = db.Column(UUID(as_uuid=False), nullable=False)
    status = db.Column(db.String(20), nullable=False)
    current_version = db.Column(db.Integer, nullable=False)
    created_by = db.Column(UUID(as_uuid=False), nullable=False)


class SemanticAssetVersion(db.Model):
    __tablename__ = "semantic_asset_versions"
    __table_args__ = (
        db.UniqueConstraint("asset_uid", "version"),
        {"schema": "public"},
    )

    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
    asset_uid = db.Column(UUID(as_uuid=False), nullable=False)
    version = db.Column(db.Integer, nullable=False)
    status = db.Column(db.String(20), nullable=False)
    content_hash = db.Column(db.String(64), nullable=False)
    definition = db.Column(JSONB, nullable=False)
    change_reason = db.Column(db.String(1000), nullable=False)
    created_by = db.Column(UUID(as_uuid=False), nullable=False)
    rollback_from_version = db.Column(db.Integer)


class SemanticAssetReview(db.Model):
    __tablename__ = "semantic_asset_reviews"
    __table_args__ = (
        db.UniqueConstraint("asset_uid", "version"),
        {"schema": "public"},
    )

    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
    asset_uid = db.Column(UUID(as_uuid=False), nullable=False)
    version = db.Column(db.Integer, nullable=False)
    decision = db.Column(db.String(20), nullable=False)
    reason = db.Column(db.String(1000), nullable=False)
    reviewer_uid = db.Column(UUID(as_uuid=False), nullable=False)


class SemanticAssetLink(db.Model):
    __tablename__ = "semantic_asset_links"
    __table_args__ = {"schema": "public"}

    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
    asset_uid = db.Column(UUID(as_uuid=False), nullable=False)
    version = db.Column(db.Integer, nullable=False)
    link_type = db.Column(db.String(30), nullable=False)
    target_type = db.Column(db.String(60), nullable=False)
    target_uid = db.Column(db.String(200), nullable=False)
    target_key = db.Column(db.String(200))
    metadata_json = db.Column("metadata", JSONB, nullable=False)


class DataElementFieldMapping(db.Model):
    __tablename__ = "data_element_field_mappings"
    __table_args__ = (
        db.UniqueConstraint("asset_uid", "field_name", "data_element_uid"),
        {"schema": "public"},
    )

    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
    asset_uid = db.Column(UUID(as_uuid=False), nullable=False)
    field_name = db.Column(db.String(200), nullable=False)
    data_element_uid = db.Column(UUID(as_uuid=False), nullable=False)
    data_element_version = db.Column(db.Integer, nullable=False)
    owner_uid = db.Column(UUID(as_uuid=False), nullable=False)
    status = db.Column(db.String(20), nullable=False)
    evidence = db.Column(JSONB, nullable=False)
    created_by = db.Column(UUID(as_uuid=False), nullable=False)


class SemanticPublicationAudit(db.Model):
    __tablename__ = "semantic_publication_audits"
    __table_args__ = {"schema": "public"}

    uid = db.Column(UUID(as_uuid=False), primary_key=True, default=new_governance_uid)
    target_type = db.Column(db.String(40), nullable=False)
    target_uid = db.Column(UUID(as_uuid=False), nullable=False)
    target_version = db.Column(db.Integer, nullable=False)
    action = db.Column(db.String(40), nullable=False)
    actor_uid = db.Column(UUID(as_uuid=False), nullable=False)
    detail = db.Column(JSONB, nullable=False)