AI-Cognitive-Leap/agnes-the-ai-analyst
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add direct catalog URL to YAML header (metric/table entity links)

Browse source 
Source line now links directly to the entity in OpenMetadata:
- metrics: https://datacatalog.../metric/UniqueVisitors
- tables:  https://datacatalog.../table/bigquery.project.dataset.table
This commit is contained in:
Petr 2026-03-15 02:03:27 +01:00
parent ab1a93ed67
commit 60039c0af3
1 changed files with 9 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
src/catalog_export.py
View file

@ -68,21 +68,26 @@ def _get_docs_dir() -> Path:
return docs_dir return docs_dir
def _yaml_header(catalog_url: str, fqn: str = "") -> str: def _yaml_header(catalog_url: str, fqn: str = "", entity_type: str = "") -> str:
""" """
Generate AUTO-GENERATED header comment for YAML files. Generate AUTO-GENERATED header comment for YAML files.
Args: Args:
catalog_url: OpenMetadata instance URL catalog_url: OpenMetadata instance URL
fqn: Optional entity FQN for reference fqn: Optional entity FQN for reference
entity_type: Entity type for catalog URL (e.g., "metric", "table")
Returns: Returns:
Multi-line comment string Multi-line comment string
""" """
timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
if fqn and entity_type:
source_url = f"{catalog_url}/{entity_type}/{fqn}"
else:
source_url = catalog_url
lines = [ lines = [
AUTO_GENERATED_MARKER, AUTO_GENERATED_MARKER,
CATALOG_SOURCE_LINE.format(url=catalog_url), CATALOG_SOURCE_LINE.format(url=source_url),
GENERATED_LINE.format(timestamp=timestamp), GENERATED_LINE.format(timestamp=timestamp),
DO_NOT_EDIT_LINE, DO_NOT_EDIT_LINE,
] ]
@ -166,7 +171,7 @@ def export_metrics(
# Write individual metric YAML # Write individual metric YAML
file_path = cat_dir / f"{metric_name}.yml" file_path = cat_dir / f"{metric_name}.yml"
header = _yaml_header(catalog_url, fqn) header = _yaml_header(catalog_url, fqn, entity_type="metric")
# Wrap in list to match MetricParser expected format # Wrap in list to match MetricParser expected format
yaml_content = yaml.dump( yaml_content = yaml.dump(
[yaml_dict], [yaml_dict],
@ -283,7 +288,7 @@ def export_tables(
# Write table YAML # Write table YAML
file_path = tables_dir / f"{table_config.name}.yml" file_path = tables_dir / f"{table_config.name}.yml"
header = _yaml_header(catalog_url, fqn) header = _yaml_header(catalog_url, fqn, entity_type="table")
yaml_content = yaml.dump( yaml_content = yaml.dump(
yaml_dict, yaml_dict,
default_flow_style=False, default_flow_style=False,