fastflowtransform.ci.core

CiIssue dataclass

A single CI issue (error or warning) discovered during static checks.

This is intentionally generic enough to be mapped to SARIF, GitHub annotations, or plain log lines.

Source code in src/fastflowtransform/ci/core.py

@dataclass
class CiIssue:
    """
    A single CI issue (error or warning) discovered during static checks.

    This is intentionally generic enough to be mapped to SARIF, GitHub
    annotations, or plain log lines.
    """

    level: str  # "error" | "warning"
    code: str  # short machine-friendly code, e.g. "GRAPH_CYCLE"
    message: str  # human-readable
    file: str | None = None
    line: int | None = None
    column: int | None = None
    obj_name: str | None = None  # model/source/test name, if applicable

CiSummary dataclass

Overall result of a CI check run.

Source code in src/fastflowtransform/ci/core.py

@dataclass
class CiSummary:
    """
    Overall result of a CI check run.
    """

    issues: list[CiIssue]
    selected_nodes: list[str]
    all_nodes: list[str]

run_ci_check

run_ci_check(*, select=None)

Run static CI checks against the currently-loaded project.

Assumptions

• REGISTRY.load_project(...) has already been called (via _prepare_context).
• No database connections are required (purely static).

Returns:

Type	Description
CiSummary	CiSummary with issues and selection info.

Exit codes are handled by the CLI wrapper, based on the returned issues.

Source code in src/fastflowtransform/ci/core.py

def run_ci_check(
    *,
    select: Sequence[str] | None = None,
) -> CiSummary:
    """
    Run static CI checks against the currently-loaded project.

    Assumptions:
      - REGISTRY.load_project(...) has already been called (via _prepare_context).
      - No database connections are required (purely static).

    Returns:
        CiSummary with issues and selection info.

    Exit codes are handled by the CLI wrapper, based on the returned issues.
    """
    issues: list[CiIssue] = []

    # --- Gather all nodes (models) known to the registry --------------------
    nodes = getattr(REGISTRY, "nodes", {}) or {}
    all_nodes = sorted(nodes.keys())

    if not all_nodes:
        issues.append(
            CiIssue(
                level="warning",
                code="NO_MODELS",
                message="Registry contains no models for this project.",
            )
        )

    # --- Graph sanity checks (missing deps + cycles) ------------------------
    issues.extend(_graph_issues())

    # --- Selection preview ---------------------------------------------------
    selected_nodes = _resolve_selection(all_nodes, select)

    if all_nodes and not selected_nodes:
        issues.append(
            CiIssue(
                level="warning",
                code="SELECTION_EMPTY",
                message="Selection is empty: no nodes would run with the given --select.",
            )
        )

    return CiSummary(
        issues=issues,
        selected_nodes=selected_nodes,
        all_nodes=all_nodes,
    )