DataOps-platform/deployment/app/api/meta_data/domain_templates.py

"""Management API for generic governance domain templates."""

from __future__ import annotations

from flask import g, jsonify, request

from app import db
from app.api.meta_data import bp
from app.core.governance.domain_template_repository import (
    SqlAlchemyDomainTemplateRepository,
)
from app.core.governance.domain_templates import (
    DomainTemplateNotFound,
    DomainTemplateService,
    DomainTemplateValidationError,
)
from app.models.result import failed, success


def _service():
    return DomainTemplateService(SqlAlchemyDomainTemplateRepository(db.session))


@bp.route("/domain-templates", methods=["GET"])
def list_domain_templates():
    return jsonify(success(_service().list_templates()))


@bp.route("/domain-templates/<template_code>", methods=["GET"])
def get_domain_template(template_code):
    try:
        return jsonify(success(_service().get_template(template_code)))
    except DomainTemplateNotFound as exc:
        return jsonify(failed(str(exc), code=404)), 404
    except DomainTemplateValidationError as exc:
        return jsonify(failed(str(exc), code=400)), 400


@bp.route("/domain-templates/<template_code>/imports", methods=["GET"])
def list_domain_template_imports(template_code):
    try:
        return jsonify(success(_service().list_imports(template_code)))
    except DomainTemplateValidationError as exc:
        return jsonify(failed(str(exc), code=400)), 400


@bp.route("/domain-templates/dry-run", methods=["POST"])
def dry_run_domain_template():
    try:
        return jsonify(success(_service().dry_run(request.get_json(silent=True))))
    except DomainTemplateValidationError as exc:
        return jsonify(failed(str(exc), code=400)), 400


@bp.route("/domain-templates/import", methods=["POST"])
def import_domain_template():
    try:
        result = _service().import_template(
            request.get_json(silent=True),
            actor_uid=g.current_user["id"],
        )
        db.session.commit()
        return jsonify(success(result, "领域模板已导入"))
    except DomainTemplateValidationError as exc:
        db.session.rollback()
        return jsonify(failed(str(exc), code=400)), 400
    except Exception:
        db.session.rollback()
        raise

@bp.route("/domain-templates/<template_code>/rollback", methods=["POST"])
def rollback_domain_template(template_code):
    try:
        body = request.get_json(silent=True) or {}
        target_version = body.get("target_version")
        if isinstance(target_version, bool):
            target_version = None
        if isinstance(target_version, str) and target_version.isdigit():
            target_version = int(target_version)
        result = _service().rollback(
            template_code,
            target_version=target_version,
            actor_uid=g.current_user["id"],
        )
        db.session.commit()
        return jsonify(success(result, "领域模板已回滚为新版本"))
    except DomainTemplateNotFound as exc:
        db.session.rollback()
        return jsonify(failed(str(exc), code=404)), 404
    except DomainTemplateValidationError as exc:
        db.session.rollback()
        return jsonify(failed(str(exc), code=400)), 400
    except Exception:
        db.session.rollback()
        raise