fastflowtransform.docs

read_docs_metadata

read_docs_metadata(project_dir)

Merge YAML + Markdown descriptions with priority: Markdown > YAML. Returns: { "models": { : { "description_html": "

…

" | None, "columns": { : "

…

" } }, }, "columns": { : { : "

…

" } } }

Source code in src/fastflowtransform/docs.py

def read_docs_metadata(project_dir: Path) -> dict[str, Any]:
    """
    Merge YAML + Markdown descriptions with priority: Markdown > YAML.
    Returns:
      {
        "models": {
          <model>: {
            "description_html": "<p>…</p>" | None,
            "columns": { <col>: "<p>…</p>" }
          },
        },
        "columns": { <relation>: { <col>: "<p>…</p>" } }
      }
    """
    # 1) YAML (from project.yml → docs.models)
    yaml_models = _read_project_yaml_docs(project_dir)  # {model: {description, columns{}}}
    out_models: dict[str, dict[str, Any]] = {}
    for model, meta in yaml_models.items() if isinstance(yaml_models, dict) else []:
        desc = (meta or {}).get("description")
        cols = (meta or {}).get("columns") or {}
        lineage_yaml = (meta or {}).get("lineage")

        out_models[model] = {
            "description_html": _render_minimarkdown(desc) if desc else None,
            "columns": {
                str(k): _render_minimarkdown(str(v))
                for k, v in (cols.items() if isinstance(cols, dict) else [])
            },
        }
        if isinstance(lineage_yaml, dict):
            out_models[model]["lineage"] = lineage_yaml

    # 2) Markdown model overrides: docs/models/<model>.md
    md_models_dir = project_dir / "docs" / "models"
    if md_models_dir.exists():
        for p in md_models_dir.glob("*.md"):
            model_name = p.stem
            _, body = _read_markdown_file(p)
            if body.strip():
                out_models.setdefault(model_name, {"description_html": None, "columns": {}})
                out_models[model_name]["description_html"] = _render_minimarkdown(body)

    # 3) Markdown column overrides: docs/columns/<relation>/<column>.md
    out_columns: dict[str, dict[str, str]] = {}
    cols_root = project_dir / "docs" / "columns"
    if cols_root.exists():
        for rel_dir in cols_root.iterdir():
            if not rel_dir.is_dir():
                continue
            rel = rel_dir.name
            for p in rel_dir.glob("*.md"):
                col = p.stem
                _, body = _read_markdown_file(p)
                if body.strip():
                    out_columns.setdefault(rel, {})[col] = _render_minimarkdown(body)

    return {"models": out_models, "columns": out_columns}