#!/usr/bin/env python3 """Downstream-first live metadata exporter.""" from __future__ import annotations import argparse import json import re import shutil import subprocess from dataclasses import dataclass, field from pathlib import Path from typing import Any from urllib.parse import parse_qs, unquote, urlparse DEFAULT_TABLES = [ "_System_Object", "_System_Field", "_System_RecordType", "_System_ValidationRule", "_System_PageLayout", "_System_Permission", "_System_ObjectFieldMapping", "_System_ReportType", "_System_Relationship", "FlowDefinition", ] PACK_TABLES = DEFAULT_TABLES TABLE_ORDER_BY = { "_System_Object": ["ApiName", "Id"], "_System_Field": ["ObjectId", "ApiName", "Id"], "_System_RecordType": ["ObjectId", "Name", "Id"], "_System_ValidationRule": ["ObjectApiName", "Name", "Id"], "_System_PageLayout": ["ObjectApiName", "LayoutName", "Id"], "_System_Permission": ["ApiName", "Id"], "_System_ObjectFieldMapping": ["TargetObject", "MappingActionType", "TargetFieldAPIName", "Id"], "_System_ReportType": ["FullName", "Id"], "_System_Relationship": ["ChildObjectApiName", "FieldApiName", "Id"], "FlowDefinition": ["DeveloperName", "Id"], } VOLATILE_FIELDS = { "CreatedDate", "LastModifiedDate", "LastModifiedById", "CreatedById", "GenerationVersion", "LayoutVersion", } @dataclass class ConnectionConfig: username: str password: str host: str port: int database: str tls: bool @dataclass class ColumnInfo: name: str data_type: str @dataclass class ComponentIndexEntry: kind: str business_key: str path: str layer_hint: str signals: list[str] = field(default_factory=list) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Export live DB metadata into a diff-friendly snapshot.", ) parser.add_argument("--dsn", required=True, help="Database DSN.") parser.add_argument("--output-dir", required=True, help="Directory to write export output into.") parser.add_argument( "--mode", choices=["raw", "pack"], default="raw", help="Export mode. raw=metadata tables, pack=component-shaped export for diffing.", ) parser.add_argument( "--tables", nargs="*", default=DEFAULT_TABLES, help="Metadata tables to export (raw mode only).", ) return parser.parse_args() def write_json(path: Path, payload: object) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2, sort_keys=True, ensure_ascii=False) + "\n", encoding="utf-8") def parse_dsn(dsn: str) -> ConnectionConfig: parsed = urlparse(dsn) if parsed.scheme != "mysql": raise ValueError(f"unsupported DSN scheme: {parsed.scheme}") if not parsed.hostname: raise ValueError("DSN must include a host") if not parsed.path or parsed.path == "/": raise ValueError("DSN must include a database name") host = parsed.hostname if host.startswith("tcp(") and host.endswith(")"): host = host[4:-1] query = parse_qs(parsed.query) tls = query.get("tls", ["false"])[0].lower() in {"1", "true", "required"} return ConnectionConfig( username=unquote(parsed.username or ""), password=unquote(parsed.password or ""), host=host, port=parsed.port or 3306, database=parsed.path.lstrip("/"), tls=tls, ) def resolve_mysql_binary() -> str: for candidate in ("mysql", "mariadb"): path = shutil.which(candidate) if path: return path raise RuntimeError("mysql or mariadb client not found in PATH") def build_mysql_cmd(config: ConnectionConfig, sql: str) -> list[str]: binary = resolve_mysql_binary() user = config.username if config.host not in {"127.0.0.1", "localhost"} and not user.endswith(".root"): user = user + ".root" cmd = [ binary, "--batch", "--raw", "--skip-column-names", "--host", config.host, "--port", str(config.port), "--user", user, f"--password={config.password}", config.database, "-e", sql, ] if config.tls: if Path(binary).name == "mysql": cmd.insert(1, "--ssl-mode=REQUIRED") else: cmd.insert(1, "--ssl") return cmd def mysql_query(config: ConnectionConfig, sql: str) -> list[str]: cmd = build_mysql_cmd(config, sql) result = subprocess.run(cmd, check=True, text=True, capture_output=True) return [line for line in result.stdout.splitlines() if line.strip()] def list_columns(config: ConnectionConfig, table: str) -> list[ColumnInfo]: rows = mysql_query( config, ( "SELECT COLUMN_NAME, DATA_TYPE " "FROM information_schema.columns " f"WHERE table_schema = '{sql_quote(config.database)}' " f"AND table_name = '{sql_quote(table)}' " "ORDER BY ORDINAL_POSITION" ), ) columns: list[ColumnInfo] = [] for row in rows: name, data_type = row.split("\t", 1) columns.append(ColumnInfo(name=name, data_type=data_type.lower())) return columns def build_select_sql(table: str, columns: list[ColumnInfo]) -> str: order_by = TABLE_ORDER_BY.get(table, ["Id"]) existing_order_columns = [column for column in order_by if any(info.name == column for info in columns)] order_clause = "" if existing_order_columns: order_clause = " ORDER BY " + ", ".join(f"`{column}`" for column in existing_order_columns) json_object_args = ", ".join( f"'{column.name}', `{column.name}`" for column in columns ) return "SELECT JSON_OBJECT(" + json_object_args + ")" + f" FROM `{table}`" + order_clause def export_table(config: ConnectionConfig, table: str) -> list[dict[str, object]]: columns = list_columns(config, table) if not columns: raise RuntimeError(f"table not found or has no columns: {table}") rows = mysql_query(config, build_select_sql(table, columns)) return [json.loads(row) for row in rows] def sql_quote(value: str) -> str: return value.replace("\\", "\\\\").replace("'", "\\'") def redact_dsn(dsn: str) -> str: if "@" not in dsn: return dsn prefix, suffix = dsn.split("@", 1) if "://" in prefix: scheme, _rest = prefix.split("://", 1) return f"{scheme}://***@{suffix}" return f"***@{suffix}" def normalize_jsonish(value: Any) -> Any: if isinstance(value, str): text = value.strip() if text and text[0] in "[{": try: return normalize_jsonish(json.loads(text)) except json.JSONDecodeError: return value return value if isinstance(value, list): normalized = [normalize_jsonish(item) for item in value] if all(isinstance(item, dict) for item in normalized): return sorted(normalized, key=business_sort_key) return normalized if isinstance(value, dict): return {key: normalize_jsonish(val) for key, val in sorted(value.items())} return value def compact_row(row: dict[str, Any], drop_fields: set[str] | None = None) -> dict[str, Any]: drop = set(drop_fields or set()) | VOLATILE_FIELDS result: dict[str, Any] = {} for key, value in row.items(): if key in drop: continue if value in ("", None, [], {}): continue result[key] = normalize_jsonish(value) return result def business_sort_key(item: dict[str, Any]) -> tuple: for key in ( "apiName", "objectApiName", "name", "fullName", "layoutName", "DeveloperName", "Id", ): if key in item: return (str(item.get(key, "")), json.dumps(item, sort_keys=True, ensure_ascii=False)) return (json.dumps(item, sort_keys=True, ensure_ascii=False),) def slugify_layout_name(name: str) -> str: slug = re.sub(r"[^a-z0-9]+", "_", name.lower()).strip("_") return slug or "layout" def layer_hint(signals: list[str]) -> str: if any(signal in {"isCustomized", "ownerPackMissing", "sourceAdmin", "sourceCloned", "sourceCurated"} for signal in signals): return "likely_l2_admin" if any(signal in {"ownerPackPresent", "sourceAutoGenerated"} for signal in signals): return "pack_owned_or_l1" return "unknown" def append_index(index: list[ComponentIndexEntry], kind: str, business_key: str, path: str, signals: list[str]) -> None: index.append( ComponentIndexEntry( kind=kind, business_key=business_key, path=path, layer_hint=layer_hint(signals), signals=sorted(set(signals)), ) ) _SF_TYPE_MAP = { "varchar": "string", "char": "string", "text": "textarea", "longtext": "textarea", "mediumtext": "textarea", "tinytext": "textarea", "int": "int", "bigint": "int", "smallint": "int", "mediumint": "int", "decimal": "double", "numeric": "double", "float": "double", "double": "double", "datetime": "datetime", "timestamp": "datetime", "date": "date", "time": "time", "bool": "boolean", "tinyint": "boolean", "blob": "base64", "longblob": "base64", "mediumblob": "base64", "json": "string", } def map_field_type(db_type: str, field_name: str) -> str: base_type = db_type.lower().split("(")[0].split(" ")[0] mapped = _SF_TYPE_MAP.get(base_type, "string") lowered_name = field_name.lower() if "phone" in lowered_name: return "phone" if "url" in lowered_name or "website" in lowered_name: return "url" if "email" in lowered_name: return "email" if "currency" in lowered_name or "amount" in lowered_name: return "currency" if "percent" in lowered_name: return "percent" if "textarea" in lowered_name or "description" in lowered_name: return "textarea" return mapped def export_object_specs(objects_raw: list[dict[str, Any]], fields_raw: list[dict[str, Any]], output_dir: Path, index: list[ComponentIndexEntry]) -> None: fields_by_object: dict[str, list[dict[str, Any]]] = {} for field in fields_raw: object_id = str(field.get("ObjectId") or "") if object_id: fields_by_object.setdefault(object_id, []).append(field) objects_dir = output_dir / "objects" objects_dir.mkdir(parents=True, exist_ok=True) for obj in objects_raw: object_id = str(obj.get("Id") or "") object_api_name = str(obj.get("ApiName") or "") if not object_api_name: continue fields = [] for field in sorted(fields_by_object.get(object_id, []), key=lambda item: str(item.get("ApiName") or "")): field_api_name = str(field.get("ApiName") or "") if not field_api_name: continue db_type = str(field.get("Type") or "string") if db_type == "formula": field_spec: dict[str, Any] = { "apiName": field_api_name, "label": field.get("Label") or field_api_name, "type": "formula", "formula": field.get("Formula") or "", } if field.get("ReturnType"): field_spec["returnType"] = field["ReturnType"] elif db_type == "reference": field_spec = { "apiName": field_api_name, "label": field.get("Label") or field_api_name, "type": "reference", } reference_to = normalize_jsonish(field.get("ReferenceTo")) if isinstance(reference_to, str): refs = [part.strip() for part in reference_to.split(",") if part.strip()] if refs: field_spec["referenceTo"] = refs elif isinstance(reference_to, list) and reference_to: field_spec["referenceTo"] = reference_to if field.get("RelationshipName"): field_spec["relationshipName"] = field["RelationshipName"] if field.get("ChildRelationshipName"): field_spec["childRelationshipName"] = field["ChildRelationshipName"] elif db_type == "picklist": field_spec = { "apiName": field_api_name, "label": field.get("Label") or field_api_name, "type": "picklist", } options = normalize_jsonish(field.get("Options")) if options not in (None, "", []): field_spec["options"] = options else: field_spec = { "apiName": field_api_name, "label": field.get("Label") or field_api_name, "type": map_field_type(db_type, field_api_name), } if field.get("Required") in (1, True): field_spec["required"] = True if field.get("Createable") in (0, False): field_spec["createable"] = False if field.get("Updateable") in (0, False): field_spec["updateable"] = False if field.get("IsNameField") in (1, True): field_spec["isNameField"] = True if field.get("IsPolymorphic") in (1, True): field_spec["isPolymorphic"] = True if field.get("ControllingField"): field_spec["controllingField"] = field["ControllingField"] if field.get("DefaultValue") not in (None, "", "null"): field_spec["default"] = str(field["DefaultValue"]) if field.get("WriteProtectionPolicy"): field_spec["writeProtectionPolicy"] = field["WriteProtectionPolicy"] fields.append(field_spec) object_spec: dict[str, Any] = { "apiName": object_api_name, "label": obj.get("Label") or object_api_name, "tableType": "custom_object" if object_api_name.endswith("__c") else "standard_object", "fields": fields, } for source_key, target_key in ( ("KeyPrefix", "keyPrefix"), ("PluralLabel", "pluralLabel"), ("Category", "category"), ("SharingModel", "sharingModel"), ("Description", "description"), ("AccessPolicy", "accessPolicy"), ): if obj.get(source_key): object_spec[target_key] = obj[source_key] if obj.get("EnableHierarchySharing") in (1, True): object_spec["enableHierarchySharing"] = True path = objects_dir / f"{object_api_name}.json" write_json(path, [object_spec]) signals = [] if object_api_name.endswith("__c"): signals.append("customApiName") if obj.get("OwnerPack"): signals.append("ownerPackPresent") else: signals.append("ownerPackMissing") append_index(index, "object_spec", f"object:{object_api_name}", str(path.relative_to(output_dir)), signals) def export_record_types(objects_raw: list[dict[str, Any]], record_types_raw: list[dict[str, Any]], output_dir: Path, index: list[ComponentIndexEntry]) -> None: object_name_by_id = { str(obj.get("Id") or ""): str(obj.get("ApiName") or "") for obj in objects_raw if obj.get("Id") and obj.get("ApiName") } grouped: dict[str, list[dict[str, Any]]] = {} for row in record_types_raw: object_api_name = object_name_by_id.get(str(row.get("ObjectId") or "")) if not object_api_name: continue item = { "name": row.get("Name"), } if row.get("Description") not in (None, ""): item["description"] = row["Description"] if row.get("record_type_code") not in (None, ""): item["code"] = row["record_type_code"] if row.get("IsActive") is False: item["active"] = False if row.get("IsMaster") is True: item["isMaster"] = True if row.get("OwnerPack"): item["ownerPack"] = row["OwnerPack"] grouped.setdefault(object_api_name, []).append(item) payload = { "recordTypes": [ { "objectApiName": object_api_name, "types": sorted(types, key=business_sort_key), } for object_api_name, types in sorted(grouped.items()) ] } path = output_dir / "record_types" / "record_types.json" write_json(path, payload) for row in payload["recordTypes"]: signals = [] if row["objectApiName"].endswith("__c"): signals.append("customApiName") if any("ownerPack" in item for item in row["types"]): signals.append("ownerPackPresent") else: signals.append("ownerPackMissing") append_index(index, "record_type", f"record_type:{row['objectApiName']}", str(path.relative_to(output_dir)), signals) def export_validation_rules(validation_raw: list[dict[str, Any]], output_dir: Path, index: list[ComponentIndexEntry]) -> None: rules = [] for row in validation_raw: item = { "id": row.get("Id"), "objectApiName": row.get("ObjectApiName"), "name": row.get("Name"), "description": row.get("Description") or "", "active": bool(row.get("Active")), "condition": row.get("Condition") or "", "errorMessage": row.get("ErrorMessage") or "", } if row.get("ErrorDisplayField"): item["errorDisplayField"] = row["ErrorDisplayField"] bindings = normalize_jsonish(row.get("RecordSetBindings")) if bindings not in (None, "", [], {}): item["recordSetBindings"] = bindings if row.get("OwnerPack"): item["ownerPack"] = row["OwnerPack"] if row.get("IsCustomized") is True: item["isCustomized"] = True rules.append(item) payload = {"validationRules": sorted(rules, key=business_sort_key)} path = output_dir / "validation" / "standard_validation_rules.json" write_json(path, payload) for rule in payload["validationRules"]: signals = [] if rule.get("objectApiName", "").endswith("__c"): signals.append("customApiName") if rule.get("isCustomized"): signals.append("isCustomized") if rule.get("ownerPack"): signals.append("ownerPackPresent") else: signals.append("ownerPackMissing") append_index( index, "validation_rule", f"validation_rule:{rule.get('objectApiName')}:{rule.get('name')}", str(path.relative_to(output_dir)), signals, ) def export_permissions(permission_raw: list[dict[str, Any]], output_dir: Path, index: list[ComponentIndexEntry]) -> None: permissions = [] for row in permission_raw: item = { "apiName": row.get("ApiName"), "label": row.get("Label"), "category": row.get("Category"), } if row.get("Description"): item["description"] = row["Description"] if row.get("RequiresPermission"): item["requiresPermission"] = row["RequiresPermission"] if row.get("IsSystem") is True: item["isSystem"] = True if row.get("IsActive") is False: item["isActive"] = False if row.get("OwnerPack"): item["ownerPack"] = row["OwnerPack"] permissions.append(item) permissions.sort(key=business_sort_key) path = output_dir / "permissions" / "crm_standard_permissions.json" write_json(path, permissions) for permission in permissions: signals = ["ownerPackPresent" if permission.get("ownerPack") else "ownerPackMissing"] append_index( index, "permission", f"permission:{permission.get('apiName')}", str(path.relative_to(output_dir)), signals, ) def export_report_types(report_type_raw: list[dict[str, Any]], output_dir: Path, index: list[ComponentIndexEntry]) -> None: report_types = [] for row in report_type_raw: item = { "fullName": row.get("FullName"), "label": row.get("Label"), "description": row.get("Description") or "", "baseObject": row.get("BaseObject"), "deployed": bool(row.get("Deployed")), "autogenerated": bool(row.get("Autogenerated")), } for key, source in ( ("sections", "Sections"), ("categories", "Categories"), ("join", "Join"), ("metadata", "Metadata"), ): normalized = normalize_jsonish(row.get(source)) if normalized not in (None, "", [], {}): item[key] = normalized if row.get("OwnerPack"): item["ownerPack"] = row["OwnerPack"] report_types.append(item) payload = {"reportTypes": sorted(report_types, key=business_sort_key)} path = output_dir / "report_types" / "standard_report_types.json" write_json(path, payload) for report_type in payload["reportTypes"]: signals = [] if str(report_type.get("baseObject") or "").endswith("__c"): signals.append("customApiName") if report_type.get("ownerPack"): signals.append("ownerPackPresent") else: signals.append("ownerPackMissing") append_index( index, "report_type", f"report_type:{report_type.get('fullName')}", str(path.relative_to(output_dir)), signals, ) def export_layouts(layout_raw: list[dict[str, Any]], output_dir: Path, index: list[ComponentIndexEntry]) -> None: layouts_dir = output_dir / "layouts" layouts_dir.mkdir(parents=True, exist_ok=True) for row in layout_raw: object_api_name = str(row.get("ObjectApiName") or "") layout_name = str(row.get("LayoutName") or "") if not object_api_name or not layout_name: continue config = normalize_jsonish(row.get("Config")) or {} if not isinstance(config, dict): config = {"config": config} payload: dict[str, Any] = dict(config) payload["objectApiName"] = object_api_name payload["layoutName"] = layout_name if row.get("Type"): payload["type"] = row["Type"] if row.get("IsDefault") is True: payload["isDefault"] = True if row.get("IsCustomized") is True: payload["isCustomized"] = True if row.get("Source"): payload["source"] = row["Source"] file_name = f"{slugify_layout_name(object_api_name)}_{slugify_layout_name(layout_name)}.json" path = layouts_dir / file_name write_json(path, payload) signals = [] source = str(row.get("Source") or "") if row.get("IsCustomized") is True: signals.append("isCustomized") if source == "admin_created": signals.append("sourceAdmin") elif source == "cloned": signals.append("sourceCloned") elif source == "curated": signals.append("sourceCurated") elif source == "auto_generated": signals.append("sourceAutoGenerated") if object_api_name.endswith("__c"): signals.append("customApiName") append_index( index, "page_layout", f"page_layout:{object_api_name}:{layout_name}", str(path.relative_to(output_dir)), signals, ) def export_relationships(relationship_raw: list[dict[str, Any]], output_dir: Path, index: list[ComponentIndexEntry]) -> None: items = [] for row in relationship_raw: item = compact_row(row) item["relatedListFields"] = normalize_jsonish(item.get("RelatedListFields")) items.append(item) items.sort(key=business_sort_key) path = output_dir / "relationships" / "standard_relationships.json" write_json(path, {"relationships": items}) for item in items: signals = [] if str(item.get("ChildObjectApiName") or "").endswith("__c") or str(item.get("ParentObjectApiName") or "").endswith("__c"): signals.append("customApiName") append_index( index, "relationship", f"relationship:{item.get('ChildObjectApiName')}:{item.get('FieldApiName')}", str(path.relative_to(output_dir)), signals, ) def export_field_mappings(mapping_raw: list[dict[str, Any]], output_dir: Path, index: list[ComponentIndexEntry]) -> None: items = [compact_row(row) for row in mapping_raw] items.sort(key=business_sort_key) path = output_dir / "field_mappings" / "object_field_mappings.json" write_json(path, {"fieldMappings": items}) for item in items: signals = [] if str(item.get("TargetObject") or "").endswith("__c"): signals.append("customApiName") append_index( index, "field_mapping", f"field_mapping:{item.get('TargetObject')}:{item.get('MappingActionType')}:{item.get('TargetFieldAPIName')}", str(path.relative_to(output_dir)), signals, ) def export_flow_definitions(flow_definition_raw: list[dict[str, Any]], output_dir: Path, index: list[ComponentIndexEntry]) -> None: definitions = [] for row in flow_definition_raw: item = compact_row( row, drop_fields={"ActiveVersionId", "LatestVersionId", "OwnerId"}, ) item = { "developerName": item.get("DeveloperName"), "masterLabel": item.get("MasterLabel"), "description": item.get("Description", ""), "processType": item.get("ProcessType"), "triggerType": item.get("TriggerType"), "triggerObjectOrEvent": item.get("TriggerObjectOrEvent"), "isCustomized": bool(item.get("IsCustomized")), } definitions.append({key: value for key, value in item.items() if value not in (None, "", False)}) definitions.sort(key=business_sort_key) path = output_dir / "flow_definitions" / "flow_definitions.json" write_json(path, {"flowDefinitions": definitions}) for item in definitions: signals = [] if item.get("isCustomized"): signals.append("isCustomized") trigger_target = str(item.get("triggerObjectOrEvent") or "") if trigger_target.endswith("__c"): signals.append("customApiName") append_index( index, "flow_definition", f"flow_definition:{item.get('developerName')}", str(path.relative_to(output_dir)), signals, ) def export_pack_shape(config: ConnectionConfig, output_dir: Path) -> None: raw = {table: export_table(config, table) for table in PACK_TABLES} index: list[ComponentIndexEntry] = [] export_object_specs(raw["_System_Object"], raw["_System_Field"], output_dir, index) export_record_types(raw["_System_Object"], raw["_System_RecordType"], output_dir, index) export_validation_rules(raw["_System_ValidationRule"], output_dir, index) export_permissions(raw["_System_Permission"], output_dir, index) export_report_types(raw["_System_ReportType"], output_dir, index) export_layouts(raw["_System_PageLayout"], output_dir, index) export_relationships(raw["_System_Relationship"], output_dir, index) export_field_mappings(raw["_System_ObjectFieldMapping"], output_dir, index) export_flow_definitions(raw["FlowDefinition"], output_dir, index) write_json( output_dir / ".component-index.json", [ { "kind": entry.kind, "businessKey": entry.business_key, "path": entry.path, "layerHint": entry.layer_hint, "signals": entry.signals, } for entry in sorted(index, key=lambda item: (item.kind, item.business_key, item.path)) ], ) write_json( output_dir / ".export-metadata.json", { "exporter": "packs/metadata-exporter/exporter.py", "mode": "pack", "dsn_redacted": redact_dsn(f"{config.username}:***@{config.host}:{config.port}/{config.database}"), "tables": PACK_TABLES, "componentCount": len(index), }, ) print(f"wrote component-shaped export to {output_dir}") def export_metadata_snapshot(output_dir: Path, dsn: str, tables: list[str]) -> None: config = parse_dsn(dsn) metadata_dir = output_dir / "metadata" metadata_dir.mkdir(parents=True, exist_ok=True) write_json( output_dir / ".export-metadata.json", { "exporter": "packs/metadata-exporter/exporter.py", "mode": "raw-metadata", "dsn_redacted": redact_dsn(dsn), "tables": tables, }, ) for table in tables: write_json(metadata_dir / f"{table}.json", export_table(config, table)) def main() -> int: args = parse_args() output_dir = Path(args.output_dir).expanduser().resolve() output_dir.mkdir(parents=True, exist_ok=True) if args.mode == "pack": export_pack_shape(parse_dsn(args.dsn), output_dir) else: export_metadata_snapshot(output_dir, args.dsn, args.tables) print(f"export complete: {output_dir}") return 0 if __name__ == "__main__": raise SystemExit(main())