774 lines
31 KiB
Python
774 lines
31 KiB
Python
import os
|
|
import re
|
|
import json
|
|
import yaml
|
|
import glob
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import Dict, Any, List, Optional
|
|
from app.core.config import settings
|
|
from app.services.metadb_service import save_connection_to_metadb
|
|
from app.services.dbt_service import sync_dbt_models_from_nodes
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def parse_yaml_to_pipeline_dict(cfg: dict, filename: str) -> dict:
|
|
"""Parse an Airflow YAML config file into UI Canvas compatible node/edge structure."""
|
|
pipeline_id = cfg.get("pipeline_id") or filename.replace(".yaml", "").replace(".yml", "")
|
|
description = cfg.get("description", "Airflow Configured Pipeline")
|
|
schedule = cfg.get("schedule", "0 */1 * * *")
|
|
mode = cfg.get("mode", "medallion")
|
|
|
|
nodes = []
|
|
edges = []
|
|
|
|
if mode == "flexible" or "tasks" in cfg:
|
|
tasks = cfg.get("tasks", [])
|
|
for idx, t in enumerate(tasks):
|
|
node_id = str(t.get("task_id", f"node_{idx}"))
|
|
n_type = t.get("type") or t.get("operator_type") or "source_db"
|
|
label = t.get("label") or node_id
|
|
nodes.append({
|
|
"id": node_id,
|
|
"type": "customNode",
|
|
"position": {"x": 80 + idx * 260, "y": 160},
|
|
"data": {
|
|
"label": label,
|
|
"type": n_type,
|
|
"configured": True,
|
|
"config": t.get("config", {})
|
|
}
|
|
})
|
|
for dep_id in t.get("dependencies", []):
|
|
edges.append({
|
|
"id": f"e_{dep_id}_{node_id}",
|
|
"source": str(dep_id),
|
|
"target": str(node_id)
|
|
})
|
|
else:
|
|
# Medallion mode parsing
|
|
bronze = cfg.get("bronze", {})
|
|
silver = cfg.get("silver", {})
|
|
gold = cfg.get("gold", {})
|
|
|
|
src_type = bronze.get("source_type", "database")
|
|
node_src_type = "source_db"
|
|
if src_type == "gws":
|
|
node_src_type = "source_gws"
|
|
elif src_type == "api":
|
|
node_src_type = "source_api"
|
|
|
|
nodes.append({
|
|
"id": "node-bronze-src",
|
|
"type": "customNode",
|
|
"position": {"x": 80, "y": 160},
|
|
"data": {
|
|
"label": f"Bronze {src_type.upper()} Source",
|
|
"type": node_src_type,
|
|
"configured": True,
|
|
"config": bronze
|
|
}
|
|
})
|
|
|
|
nodes.append({
|
|
"id": "node-bronze-storage",
|
|
"type": "customNode",
|
|
"position": {"x": 340, "y": 160},
|
|
"data": {
|
|
"label": "MinIO Bronze Storage",
|
|
"type": "minio_storage",
|
|
"configured": True,
|
|
"config": bronze.get("target", {})
|
|
}
|
|
})
|
|
edges.append({"id": "e1-2", "source": "node-bronze-src", "target": "node-bronze-storage"})
|
|
|
|
sil_trans = silver.get("transformation", {})
|
|
if sil_trans:
|
|
nodes.append({
|
|
"id": "node-silver-dbt",
|
|
"type": "customNode",
|
|
"position": {"x": 600, "y": 160},
|
|
"data": {
|
|
"label": f"dbt Silver Model ({sil_trans.get('select_models', 'silver')})",
|
|
"type": "dbt_transform",
|
|
"configured": True,
|
|
"config": {"dbt_layer": "silver", "dbt_model": sil_trans.get("select_models", "")}
|
|
}
|
|
})
|
|
edges.append({"id": "e2-3", "source": "node-bronze-storage", "target": "node-silver-dbt"})
|
|
|
|
sil_ge = silver.get("quality_check", {})
|
|
if sil_ge:
|
|
prev_id = "node-silver-dbt" if sil_trans else "node-bronze-storage"
|
|
nodes.append({
|
|
"id": "node-silver-ge",
|
|
"type": "customNode",
|
|
"position": {"x": 860, "y": 160},
|
|
"data": {
|
|
"label": f"GE Quality ({sil_ge.get('expectation_suite', 'default')})",
|
|
"type": "ge_quality",
|
|
"configured": True,
|
|
"config": sil_ge
|
|
}
|
|
})
|
|
edges.append({"id": "e3-4", "source": prev_id, "target": "node-silver-ge"})
|
|
|
|
gold_trans = gold.get("transformation", {})
|
|
if gold_trans:
|
|
prev_id = "node-silver-ge" if sil_ge else ("node-silver-dbt" if sil_trans else "node-bronze-storage")
|
|
nodes.append({
|
|
"id": "node-gold-dbt",
|
|
"type": "customNode",
|
|
"position": {"x": 1120, "y": 160},
|
|
"data": {
|
|
"label": f"dbt Gold Mart ({gold_trans.get('select_models', 'gold')})",
|
|
"type": "dbt_transform",
|
|
"configured": True,
|
|
"config": {"dbt_layer": "gold", "dbt_model": gold_trans.get("select_models", "")}
|
|
}
|
|
})
|
|
edges.append({"id": "e4-5", "source": prev_id, "target": "node-gold-dbt"})
|
|
|
|
return {
|
|
"id": pipeline_id,
|
|
"description": description,
|
|
"schedule": schedule,
|
|
"nodes": nodes,
|
|
"edges": edges,
|
|
"is_draft": False,
|
|
"status": "deployed"
|
|
}
|
|
|
|
|
|
def get_metadata_path(filename: str) -> str:
|
|
"""Resolve existing metadata file path, checking METADATA_DIR then CONFIGS_DIR."""
|
|
meta_dir = getattr(settings, "METADATA_DIR", settings.CONFIGS_DIR)
|
|
p = os.path.join(meta_dir, filename)
|
|
if os.path.exists(p):
|
|
return p
|
|
cfg_p = os.path.join(settings.CONFIGS_DIR, filename)
|
|
if os.path.exists(cfg_p):
|
|
return cfg_p
|
|
return p
|
|
|
|
|
|
def get_target_metadata_path(filename: str) -> str:
|
|
"""Resolve target path for writing metadata file."""
|
|
meta_dir = getattr(settings, "METADATA_DIR", settings.CONFIGS_DIR)
|
|
return os.path.join(meta_dir, filename)
|
|
|
|
|
|
def get_pipeline_by_id(pipeline_id: str) -> Optional[Dict[str, Any]]:
|
|
"""Retrieve single pipeline dictionary by ID, searching JSON drafts, deployed JSON, or YAML."""
|
|
norm_id = pipeline_id.lower().replace(" ", "_")
|
|
draft_path = get_metadata_path(f"{norm_id}.draft.json")
|
|
if os.path.exists(draft_path):
|
|
try:
|
|
with open(draft_path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
data["id"] = norm_id
|
|
data["status"] = "draft"
|
|
return data
|
|
except Exception as e:
|
|
logger.warning("Error reading draft JSON %s: %s", draft_path, e)
|
|
|
|
json_path = get_metadata_path(f"{norm_id}.json")
|
|
if os.path.exists(json_path):
|
|
try:
|
|
with open(json_path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
data["id"] = norm_id
|
|
data["status"] = data.get("status", "deployed")
|
|
return data
|
|
except Exception as e:
|
|
logger.warning("Error reading metadata JSON %s: %s", json_path, e)
|
|
|
|
yaml_path = os.path.join(settings.CONFIGS_DIR, f"{norm_id}.yaml")
|
|
if os.path.exists(yaml_path):
|
|
try:
|
|
with open(yaml_path, "r", encoding="utf-8") as f:
|
|
cfg = yaml.safe_load(f)
|
|
if isinstance(cfg, dict):
|
|
return parse_yaml_to_pipeline_dict(cfg, f"{norm_id}.yaml")
|
|
except Exception as e:
|
|
logger.warning("Error reading YAML %s: %s", yaml_path, e)
|
|
|
|
return None
|
|
|
|
|
|
def list_all_pipelines() -> List[Dict[str, Any]]:
|
|
"""Retrieve all pipeline configs (JSON drafts/metadata and deployed YAMLs)."""
|
|
os.makedirs(settings.CONFIGS_DIR, exist_ok=True)
|
|
if hasattr(settings, "METADATA_DIR"):
|
|
os.makedirs(settings.METADATA_DIR, exist_ok=True)
|
|
|
|
pipelines_map: Dict[str, Dict[str, Any]] = {}
|
|
|
|
# 1. Load JSON files (saved drafts and UI canvas snapshots)
|
|
scan_dirs = []
|
|
if hasattr(settings, "METADATA_DIR") and os.path.exists(settings.METADATA_DIR):
|
|
scan_dirs.append(settings.METADATA_DIR)
|
|
container_meta = "/app/merica-pipelines/metadata"
|
|
if container_meta not in scan_dirs and os.path.exists(container_meta):
|
|
scan_dirs.append(container_meta)
|
|
if settings.CONFIGS_DIR not in scan_dirs and os.path.exists(settings.CONFIGS_DIR):
|
|
scan_dirs.append(settings.CONFIGS_DIR)
|
|
|
|
for s_dir in scan_dirs:
|
|
for file_name in sorted(os.listdir(s_dir)):
|
|
file_path = os.path.join(s_dir, file_name)
|
|
if file_name.endswith(".json") and not ".version." in file_name:
|
|
if file_name in ["rbac_mappings.json", "package.json", "tsconfig.json"]:
|
|
continue
|
|
try:
|
|
with open(file_path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
if not isinstance(data, dict):
|
|
continue
|
|
if "nodes" not in data and not file_name.endswith(".draft.json") and "tasks" not in data and "pipeline_id" not in data:
|
|
continue
|
|
p_id = data.get("id") or data.get("pipeline_id") or file_name.replace(".draft.json", "").replace(".json", "")
|
|
data["id"] = p_id
|
|
if file_name.endswith(".draft.json") or data.get("is_draft"):
|
|
data["status"] = "draft"
|
|
else:
|
|
data["status"] = "deployed"
|
|
pipelines_map[p_id] = data
|
|
except Exception as e:
|
|
logger.warning("Error reading JSON file %s: %s", file_name, e)
|
|
|
|
# 2. Load YAML files (real Airflow DAG configs)
|
|
yaml_scan_dirs = []
|
|
if os.path.exists(settings.CONFIGS_DIR):
|
|
yaml_scan_dirs.append(settings.CONFIGS_DIR)
|
|
container_cfg = "/app/merica-pipelines/configs"
|
|
if container_cfg not in yaml_scan_dirs and os.path.exists(container_cfg):
|
|
yaml_scan_dirs.append(container_cfg)
|
|
for b_dir in sorted(glob.glob("/opt/airflow/dag_bundles/*/tracking_repo/configs")):
|
|
if b_dir not in yaml_scan_dirs and os.path.exists(b_dir):
|
|
yaml_scan_dirs.append(b_dir)
|
|
|
|
for y_dir in yaml_scan_dirs:
|
|
for file_name in sorted(os.listdir(y_dir)):
|
|
file_path = os.path.join(y_dir, file_name)
|
|
if file_name.endswith(".yaml") or file_name.endswith(".yml"):
|
|
p_id = file_name.replace(".yaml", "").replace(".yml", "")
|
|
existing = pipelines_map.get(p_id)
|
|
should_load_yaml = False
|
|
if existing is None:
|
|
should_load_yaml = True
|
|
elif existing.get("status") == "deployed":
|
|
json_path = get_metadata_path(f"{p_id}.json")
|
|
if os.path.exists(json_path) and os.path.getmtime(file_path) > os.path.getmtime(json_path):
|
|
should_load_yaml = True
|
|
|
|
if should_load_yaml:
|
|
try:
|
|
with open(file_path, "r", encoding="utf-8") as f:
|
|
cfg = yaml.safe_load(f)
|
|
if isinstance(cfg, dict):
|
|
parsed_data = parse_yaml_to_pipeline_dict(cfg, file_name)
|
|
pipelines_map[p_id] = parsed_data
|
|
except Exception as e:
|
|
logger.warning("Error parsing YAML config %s: %s", file_name, e)
|
|
|
|
return list(pipelines_map.values())
|
|
|
|
|
|
def save_pipeline_draft(
|
|
pipeline_id: str,
|
|
description: Optional[str],
|
|
schedule: Optional[str],
|
|
nodes: List[Dict[str, Any]],
|
|
edges: List[Dict[str, Any]],
|
|
notes: Optional[List[Dict[str, Any]]] = None,
|
|
owner_id: Optional[str] = None,
|
|
owner_name: Optional[str] = None
|
|
) -> Dict[str, Any]:
|
|
"""Save an in-progress pipeline design as a .draft.json file."""
|
|
norm_id = pipeline_id.lower().replace(" ", "_")
|
|
target_meta_dir = getattr(settings, "METADATA_DIR", settings.CONFIGS_DIR)
|
|
os.makedirs(target_meta_dir, exist_ok=True)
|
|
|
|
json_path = os.path.join(target_meta_dir, f"{norm_id}.draft.json")
|
|
existing = get_pipeline_by_id(norm_id) if os.path.exists(json_path) else None
|
|
resolved_owner_id = owner_id or (existing.get("owner_id") if existing else None)
|
|
resolved_owner_name = owner_name or (existing.get("owner_name") if existing else None)
|
|
resolved_notes = notes if notes is not None else (existing.get("notes", []) if existing else [])
|
|
|
|
draft_data = {
|
|
"id": norm_id,
|
|
"description": description or "Dynamic Lakehouse Pipeline Draft",
|
|
"schedule": schedule or "0 */1 * * *",
|
|
"owner_id": resolved_owner_id,
|
|
"owner_name": resolved_owner_name,
|
|
"nodes": nodes,
|
|
"edges": edges,
|
|
"notes": resolved_notes,
|
|
"is_draft": True,
|
|
"status": "draft",
|
|
"updated_at": datetime.utcnow().isoformat() + "Z"
|
|
}
|
|
with open(json_path, "w", encoding="utf-8") as f:
|
|
json.dump(draft_data, f, indent=2)
|
|
|
|
return {
|
|
"status": "success",
|
|
"pipeline_id": norm_id,
|
|
"filename": f"{norm_id}.draft.json",
|
|
"message": f"Draft pipeline '{norm_id}' saved successfully."
|
|
}
|
|
|
|
|
|
def duplicate_pipeline(
|
|
pipeline_id: str,
|
|
new_pipeline_id: str,
|
|
owner_id: Optional[str] = None,
|
|
owner_name: Optional[str] = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Duplicate a pipeline with a new ID.
|
|
Per Decision #4: The cloner becomes the owner, and the status is set to draft.
|
|
"""
|
|
source = get_pipeline_by_id(pipeline_id)
|
|
if not source:
|
|
raise ValueError(f"Source pipeline '{pipeline_id}' not found.")
|
|
|
|
norm_new_id = new_pipeline_id.lower().replace(" ", "_")
|
|
return save_pipeline_draft(
|
|
pipeline_id=norm_new_id,
|
|
description=f"Copy of {source.get('description', '')}".strip(),
|
|
schedule=source.get("schedule", "0 */1 * * *"),
|
|
nodes=source.get("nodes", []),
|
|
edges=source.get("edges", []),
|
|
notes=source.get("notes", []),
|
|
owner_id=owner_id,
|
|
owner_name=owner_name
|
|
)
|
|
|
|
|
|
def build_semantic_task_id(node: Dict[str, Any], seen_ids: Optional[set] = None) -> str:
|
|
"""Transform random ReactFlow IDs (node_179...) into semantic Airflow task IDs."""
|
|
data = node.get("data", {})
|
|
cfg = data.get("config", {})
|
|
raw_id = str(node.get("id", "task"))
|
|
|
|
if not raw_id.startswith("node_"):
|
|
candidate = re.sub(r'[^a-zA-Z0-9_.-]', '_', raw_id).lower()
|
|
if seen_ids is not None:
|
|
seen_ids.add(candidate)
|
|
return candidate
|
|
|
|
op_type = (data.get("type") or "task").lower()
|
|
label = (data.get("label") or "").lower()
|
|
|
|
layer = "bronze" if any(k in op_type for k in ["source", "bronze", "sling", "cdc", "gws", "api"]) else (
|
|
"silver" if any(k in op_type for k in ["silver", "iceberg", "duckdb", "pyiceberg"]) else (
|
|
"gold" if any(k in op_type for k in ["gold", "mart"]) else (
|
|
"quality" if any(k in op_type for k in ["ge", "quality"]) else "task"
|
|
)
|
|
)
|
|
)
|
|
|
|
entity = ""
|
|
for f in ["table_name", "target_table", "dbt_model", "dataset_name"]:
|
|
if cfg.get(f):
|
|
entity = re.sub(r'[^a-zA-Z0-9]', '', str(cfg[f]).split(".")[-1].replace("_clean", "").replace("_raw", "")).lower()
|
|
break
|
|
if not entity and label:
|
|
entity = re.sub(r'[^a-zA-Z0-9]', '_', label).strip('_').lower().replace("source_", "").replace("dbt_", "").replace("_task", "")
|
|
if not entity:
|
|
entity = op_type.replace("source_", "").replace("_transform", "")
|
|
|
|
engine = "cdc" if "cdc" in op_type or "sling" in op_type or "db" in op_type else (
|
|
"gws" if "gws" in op_type or "sheet" in op_type else (
|
|
"api" if "api" in op_type else (
|
|
"duckdb" if "duckdb" in op_type else (
|
|
"pyiceberg" if "pyiceberg" in op_type else (
|
|
"spark" if "spark" in op_type else (
|
|
"dbt" if "dbt" in op_type else "op"
|
|
)
|
|
)
|
|
)
|
|
)
|
|
)
|
|
)
|
|
|
|
s_id = f"{layer}_{entity}_{engine}_task"
|
|
s_id = re.sub(r'_+', '_', s_id).strip('_')
|
|
|
|
if seen_ids is not None:
|
|
base = s_id
|
|
seq = 2
|
|
while s_id in seen_ids:
|
|
s_id = f"{base}_{seq}"
|
|
seq += 1
|
|
seen_ids.add(s_id)
|
|
|
|
return s_id
|
|
|
|
|
|
def deploy_pipeline_dag(
|
|
pipeline_id: str,
|
|
description: Optional[str],
|
|
schedule: Optional[str],
|
|
nodes: List[Dict[str, Any]],
|
|
edges: List[Dict[str, Any]],
|
|
spec_yaml: Optional[str] = "",
|
|
notes: Optional[List[Dict[str, Any]]] = None,
|
|
owner_id: Optional[str] = None,
|
|
owner_name: Optional[str] = None
|
|
) -> Dict[str, Any]:
|
|
"""Compile and deploy pipeline: sync dbt models, persist custom connections, write JSON and YAML."""
|
|
norm_id = pipeline_id.lower().replace(" ", "_")
|
|
os.makedirs(settings.CONFIGS_DIR, exist_ok=True)
|
|
|
|
# Sync dbt SQL model files physically if dbt nodes exist
|
|
sync_dbt_models_from_nodes(nodes)
|
|
|
|
# Automatically persist custom database connections if present in nodes
|
|
for node in nodes:
|
|
cfg = node.get("data", {}).get("config", {})
|
|
if cfg.get("conn_id") == "custom_db" or cfg.get("custom_host"):
|
|
conn_id = cfg.get("custom_conn_id") or f"conn_custom_{norm_id}"
|
|
save_connection_to_metadb(
|
|
conn_id=conn_id,
|
|
conn_type=cfg.get("custom_db_type", "postgres"),
|
|
host=cfg.get("custom_host", "localhost"),
|
|
port=int(cfg.get("custom_port", 5432)),
|
|
schema=cfg.get("custom_database", "public"),
|
|
login=cfg.get("custom_user", "postgres"),
|
|
password=cfg.get("custom_password", "")
|
|
)
|
|
|
|
# Build semantic node ID mapping
|
|
seen_ids = set()
|
|
node_id_map = {}
|
|
for node in nodes:
|
|
raw_id = str(node.get("id"))
|
|
node_id_map[raw_id] = build_semantic_task_id(node, seen_ids)
|
|
|
|
# 1. Save JSON metadata for UI load/edit (marking as deployed)
|
|
json_path = get_target_metadata_path(f"{norm_id}.json")
|
|
existing = get_pipeline_by_id(norm_id) if os.path.exists(json_path) else None
|
|
resolved_owner_id = owner_id or (existing.get("owner_id") if existing else None)
|
|
resolved_owner_name = owner_name or (existing.get("owner_name") if existing else None)
|
|
resolved_notes = notes if notes is not None else (existing.get("notes", []) if existing else [])
|
|
|
|
pipeline_data = {
|
|
"id": norm_id,
|
|
"description": description or "Dynamic Lakehouse Pipeline",
|
|
"schedule": schedule or "0 */1 * * *",
|
|
"owner_id": resolved_owner_id,
|
|
"owner_name": resolved_owner_name,
|
|
"nodes": nodes,
|
|
"edges": edges,
|
|
"notes": resolved_notes,
|
|
"is_draft": False,
|
|
"status": "deployed",
|
|
"updated_at": datetime.utcnow().isoformat() + "Z"
|
|
}
|
|
with open(json_path, "w", encoding="utf-8") as f:
|
|
json.dump(pipeline_data, f, indent=2)
|
|
|
|
# Remove draft file if it existed
|
|
for draft_d in [getattr(settings, "METADATA_DIR", None), settings.CONFIGS_DIR]:
|
|
if draft_d:
|
|
d_p = os.path.join(draft_d, f"{norm_id}.draft.json")
|
|
if os.path.exists(d_p):
|
|
os.remove(d_p)
|
|
|
|
# 2. Sanitize and Save YAML specification for Airflow DAG Processor
|
|
yaml_path = os.path.join(settings.CONFIGS_DIR, f"{norm_id}.yaml")
|
|
if spec_yaml:
|
|
try:
|
|
parsed_spec = yaml.safe_load(spec_yaml)
|
|
if isinstance(parsed_spec, dict) and "tasks" in parsed_spec:
|
|
for t in parsed_spec["tasks"]:
|
|
raw_tid = str(t.get("task_id"))
|
|
if raw_tid in node_id_map:
|
|
t["task_id"] = node_id_map[raw_tid]
|
|
# Remap dependencies to semantic task IDs
|
|
t["dependencies"] = [node_id_map.get(str(d), str(d)) for d in t.get("dependencies", [])]
|
|
yaml_content = yaml.dump(parsed_spec, sort_keys=False)
|
|
else:
|
|
yaml_content = spec_yaml
|
|
except Exception:
|
|
yaml_content = spec_yaml
|
|
else:
|
|
yaml_content = yaml.dump(pipeline_data, sort_keys=False)
|
|
|
|
with open(yaml_path, "w", encoding="utf-8") as f:
|
|
f.write(yaml_content)
|
|
|
|
# 3. Auto-Register pipeline & lineage to OpenMetadata 2.0 with Dynamic Governance
|
|
try:
|
|
from app.services.openmetadata_service import sync_pipeline_to_openmetadata, auto_sync_canvas_lineage
|
|
gov_dict = {}
|
|
if "parsed_spec" in locals() and isinstance(parsed_spec, dict):
|
|
gov_dict = parsed_spec.get("governance", {})
|
|
elif isinstance(pipeline_data, dict):
|
|
gov_dict = pipeline_data.get("governance", {})
|
|
if not gov_dict.get("owner") and resolved_owner_id:
|
|
gov_dict["owner"] = resolved_owner_id
|
|
if not gov_dict.get("owner_name") and resolved_owner_name:
|
|
gov_dict["owner_name"] = resolved_owner_name
|
|
|
|
sync_pipeline_to_openmetadata(norm_id, description or "", nodes=nodes, governance=gov_dict)
|
|
auto_sync_canvas_lineage(norm_id, nodes, edges)
|
|
except Exception as e:
|
|
logger.warning("OpenMetadata Auto-Registration Note: %s", e)
|
|
|
|
return {
|
|
"status": "success",
|
|
"pipeline_id": norm_id,
|
|
"filename": f"{norm_id}.yaml",
|
|
"message": f"Pipeline '{norm_id}' successfully generated & saved to {yaml_path}"
|
|
}
|
|
|
|
|
|
def delete_pipeline_files(pipeline_id: str) -> bool:
|
|
"""Delete all artifacts (json, draft.json, yaml, version snapshots) associated with a pipeline."""
|
|
norm_id = pipeline_id.lower().replace(" ", "_")
|
|
target_dirs = []
|
|
if hasattr(settings, "METADATA_DIR") and settings.METADATA_DIR:
|
|
target_dirs.append(settings.METADATA_DIR)
|
|
if settings.CONFIGS_DIR not in target_dirs:
|
|
target_dirs.append(settings.CONFIGS_DIR)
|
|
|
|
deleted = False
|
|
for d in target_dirs:
|
|
if not os.path.exists(d):
|
|
continue
|
|
json_path = os.path.join(d, f"{norm_id}.json")
|
|
draft_path = os.path.join(d, f"{norm_id}.draft.json")
|
|
yaml_path = os.path.join(d, f"{norm_id}.yaml")
|
|
|
|
for p in (json_path, draft_path, yaml_path):
|
|
if os.path.exists(p):
|
|
os.remove(p)
|
|
deleted = True
|
|
|
|
# Delete version snapshots
|
|
for file_name in os.listdir(d):
|
|
if file_name.startswith(f"{norm_id}.version.") and file_name.endswith(".json"):
|
|
v_path = os.path.join(d, file_name)
|
|
os.remove(v_path)
|
|
deleted = True
|
|
|
|
return deleted
|
|
|
|
|
|
def publish_pipeline_version(
|
|
pipeline_id: str,
|
|
version_tag: str,
|
|
changelog: Optional[str] = "",
|
|
bundle_name: Optional[str] = "dlh-core-ingestion",
|
|
nodes: Optional[List[Dict[str, Any]]] = None,
|
|
edges: Optional[List[Dict[str, Any]]] = None
|
|
) -> Dict[str, Any]:
|
|
"""Publish a permanent semantic version snapshot for Airflow 3 DAG Bundle & UI Time Machine."""
|
|
norm_id = pipeline_id.lower().replace(" ", "_")
|
|
os.makedirs(settings.CONFIGS_DIR, exist_ok=True)
|
|
if hasattr(settings, "METADATA_DIR"):
|
|
os.makedirs(settings.METADATA_DIR, exist_ok=True)
|
|
|
|
clean_tag = version_tag.strip().lower()
|
|
if not clean_tag.startswith("v"):
|
|
clean_tag = f"v{clean_tag}"
|
|
|
|
json_path = get_metadata_path(f"{norm_id}.json")
|
|
draft_path = get_metadata_path(f"{norm_id}.draft.json")
|
|
|
|
# 1. Resolve pipeline topology
|
|
pipeline_data = {}
|
|
if nodes is not None and edges is not None:
|
|
pipeline_data = {
|
|
"id": norm_id,
|
|
"description": f"Published version {clean_tag}",
|
|
"schedule": "0 */1 * * *",
|
|
"nodes": nodes,
|
|
"edges": edges,
|
|
}
|
|
elif os.path.exists(json_path):
|
|
with open(json_path, "r", encoding="utf-8") as f:
|
|
pipeline_data = json.load(f)
|
|
elif os.path.exists(draft_path):
|
|
with open(draft_path, "r", encoding="utf-8") as f:
|
|
pipeline_data = json.load(f)
|
|
else:
|
|
pipeline_data = {
|
|
"id": norm_id,
|
|
"description": f"Published version {clean_tag}",
|
|
"schedule": "0 */1 * * *",
|
|
"nodes": [],
|
|
"edges": [],
|
|
}
|
|
|
|
# Update metadata
|
|
pipeline_data["version"] = clean_tag
|
|
pipeline_data["bundle_name"] = bundle_name or "dlh-core-ingestion"
|
|
pipeline_data["changelog"] = changelog or "Released new version via MERICA Studio"
|
|
pipeline_data["published_at"] = datetime.utcnow().isoformat() + "Z"
|
|
pipeline_data["is_draft"] = False
|
|
pipeline_data["status"] = "deployed"
|
|
|
|
# Sync dbt models
|
|
if "nodes" in pipeline_data and pipeline_data["nodes"]:
|
|
sync_dbt_models_from_nodes(pipeline_data["nodes"])
|
|
|
|
# 2. Save version snapshot file in METADATA_DIR
|
|
snapshot_filename = f"{norm_id}.version.{clean_tag}.json"
|
|
snapshot_path = get_target_metadata_path(snapshot_filename)
|
|
with open(snapshot_path, "w", encoding="utf-8") as f:
|
|
json.dump(pipeline_data, f, indent=2)
|
|
|
|
# 3. Update active JSON metadata in METADATA_DIR
|
|
target_json_path = get_target_metadata_path(f"{norm_id}.json")
|
|
with open(target_json_path, "w", encoding="utf-8") as f:
|
|
json.dump(pipeline_data, f, indent=2)
|
|
|
|
# Remove draft if exists in either dir
|
|
for draft_d in [getattr(settings, "METADATA_DIR", None), settings.CONFIGS_DIR]:
|
|
if draft_d:
|
|
d_p = os.path.join(draft_d, f"{norm_id}.draft.json")
|
|
if os.path.exists(d_p):
|
|
os.remove(d_p)
|
|
|
|
# 4. Generate/Update YAML DAG Specification for Airflow 3 in CONFIGS_DIR
|
|
yaml_path = os.path.join(settings.CONFIGS_DIR, f"{norm_id}.yaml")
|
|
yaml_header = (
|
|
f"# ============================================================\n"
|
|
f"# MERICA Studio Data Lakehouse Pipeline\n"
|
|
f"# Airflow 3 DAG Bundle: {pipeline_data['bundle_name']}\n"
|
|
f"# Pinned Version: {clean_tag} | Published: {pipeline_data['published_at']}\n"
|
|
f"# Changelog: {pipeline_data['changelog']}\n"
|
|
f"# ============================================================\n"
|
|
)
|
|
yaml_body = yaml.dump(pipeline_data, sort_keys=False)
|
|
with open(yaml_path, "w", encoding="utf-8") as f:
|
|
f.write(yaml_header + yaml_body)
|
|
|
|
return {
|
|
"status": "published",
|
|
"pipeline_id": norm_id,
|
|
"version": clean_tag,
|
|
"bundle_name": pipeline_data["bundle_name"],
|
|
"changelog": pipeline_data["changelog"],
|
|
"published_at": pipeline_data["published_at"],
|
|
"snapshot_filename": snapshot_filename,
|
|
"nodes_count": len(pipeline_data.get("nodes", [])),
|
|
"edges_count": len(pipeline_data.get("edges", [])),
|
|
"message": f"Pipeline '{norm_id}' version {clean_tag} published and pinned to bundle '{pipeline_data['bundle_name']}'."
|
|
}
|
|
|
|
|
|
def list_pipeline_versions(pipeline_id: str) -> Dict[str, Any]:
|
|
"""Retrieve all published historical version snapshots for a pipeline."""
|
|
norm_id = pipeline_id.lower().replace(" ", "_")
|
|
os.makedirs(settings.CONFIGS_DIR, exist_ok=True)
|
|
if hasattr(settings, "METADATA_DIR"):
|
|
os.makedirs(settings.METADATA_DIR, exist_ok=True)
|
|
|
|
current_version = None
|
|
current_bundle = "dlh-core-ingestion"
|
|
json_path = get_metadata_path(f"{norm_id}.json")
|
|
if os.path.exists(json_path):
|
|
try:
|
|
with open(json_path, "r", encoding="utf-8") as f:
|
|
cur_data = json.load(f)
|
|
current_version = cur_data.get("version")
|
|
current_bundle = cur_data.get("bundle_name", "dlh-core-ingestion")
|
|
except Exception:
|
|
pass
|
|
|
|
versions = []
|
|
prefix = f"{norm_id}.version."
|
|
seen_snapshots = set()
|
|
scan_dirs = []
|
|
if hasattr(settings, "METADATA_DIR") and settings.METADATA_DIR:
|
|
scan_dirs.append(settings.METADATA_DIR)
|
|
if settings.CONFIGS_DIR not in scan_dirs:
|
|
scan_dirs.append(settings.CONFIGS_DIR)
|
|
|
|
for s_dir in scan_dirs:
|
|
if not os.path.exists(s_dir):
|
|
continue
|
|
for file_name in sorted(os.listdir(s_dir)):
|
|
if file_name.startswith(prefix) and file_name.endswith(".json") and file_name not in seen_snapshots:
|
|
seen_snapshots.add(file_name)
|
|
file_path = os.path.join(s_dir, file_name)
|
|
try:
|
|
with open(file_path, "r", encoding="utf-8") as f:
|
|
snap = json.load(f)
|
|
versions.append({
|
|
"version": snap.get("version", "v1.0.0"),
|
|
"pipeline_id": norm_id,
|
|
"bundle_name": snap.get("bundle_name", current_bundle),
|
|
"changelog": snap.get("changelog", ""),
|
|
"published_at": snap.get("published_at", ""),
|
|
"nodes_count": len(snap.get("nodes", [])),
|
|
"edges_count": len(snap.get("edges", [])),
|
|
"snapshot_filename": file_name
|
|
})
|
|
except Exception as e:
|
|
logger.warning("Error reading version snapshot %s: %s", file_name, e)
|
|
|
|
# Sort newest published first
|
|
versions.sort(key=lambda x: x.get("published_at", ""), reverse=True)
|
|
|
|
return {
|
|
"pipeline_id": norm_id,
|
|
"current_version": current_version,
|
|
"bundle_name": current_bundle,
|
|
"versions": versions
|
|
}
|
|
|
|
|
|
def revert_pipeline_version(pipeline_id: str, version_tag: str) -> Dict[str, Any]:
|
|
"""Revert the active canvas pipeline to a specific past historical version snapshot."""
|
|
norm_id = pipeline_id.lower().replace(" ", "_")
|
|
clean_tag = version_tag.strip().lower()
|
|
if not clean_tag.startswith("v"):
|
|
clean_tag = f"v{clean_tag}"
|
|
|
|
snapshot_filename = f"{norm_id}.version.{clean_tag}.json"
|
|
snapshot_path = get_metadata_path(snapshot_filename)
|
|
if not os.path.exists(snapshot_path):
|
|
return {
|
|
"status": "error",
|
|
"message": f"Version snapshot '{snapshot_filename}' not found for pipeline '{norm_id}'."
|
|
}
|
|
|
|
with open(snapshot_path, "r", encoding="utf-8") as f:
|
|
reverted_data = json.load(f)
|
|
|
|
# Update active JSON file in METADATA_DIR
|
|
target_json_path = get_target_metadata_path(f"{norm_id}.json")
|
|
with open(target_json_path, "w", encoding="utf-8") as f:
|
|
json.dump(reverted_data, f, indent=2)
|
|
|
|
# Recompile active YAML DAG file in CONFIGS_DIR
|
|
yaml_path = os.path.join(settings.CONFIGS_DIR, f"{norm_id}.yaml")
|
|
yaml_header = (
|
|
f"# ============================================================\n"
|
|
f"# MERICA Studio Data Lakehouse Pipeline (Reverted)\n"
|
|
f"# Airflow 3 DAG Bundle: {reverted_data.get('bundle_name', 'dlh-core-ingestion')}\n"
|
|
f"# Pinned Version: {clean_tag} | Reverted at: {datetime.utcnow().isoformat()}Z\n"
|
|
f"# ============================================================\n"
|
|
)
|
|
yaml_body = yaml.dump(reverted_data, sort_keys=False)
|
|
with open(yaml_path, "w", encoding="utf-8") as f:
|
|
f.write(yaml_header + yaml_body)
|
|
|
|
return {
|
|
"status": "reverted",
|
|
"pipeline_id": norm_id,
|
|
"active_version": clean_tag,
|
|
"pipeline_data": reverted_data,
|
|
"message": f"Successfully reverted pipeline '{norm_id}' to version {clean_tag}."
|
|
}
|
|
|