Skip to content

fastflowtransform.cli.ci_cmd

ci_check 

ci_check(project='.', env_name='dev', engine=None, vars=None, select=None)

Static CI check: parse project, validate DAG, and preview selection.

Runs without a database connection by default. Intended for PR/CI jobs:

  • Validates that models parse and dependencies are resolvable.
  • Detects dependency cycles.
  • Performs a dry-run selection based on --select (no execution).
  • Returns exit code 1 if any error-level issues are present.
Source code in src/fastflowtransform/cli/ci_cmd.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def ci_check(
    project: ProjectArg = ".",
    env_name: EnvOpt = "dev",
    engine: EngineOpt = None,
    vars: VarsOpt = None,
    select: SelectOpt = None,
) -> None:
    """
    Static CI check: parse project, validate DAG, and preview selection.

    Runs **without a database connection** by default. Intended for PR/CI jobs:

      - Validates that models parse and dependencies are resolvable.
      - Detects dependency cycles.
      - Performs a dry-run selection based on --select (no execution).
      - Returns exit code 1 if any error-level issues are present.
    """
    # Load project + registry, but do NOT create an executor → no DB work.
    ctx = _prepare_context(project, env_name, engine, vars)
    bind_context(engine=ctx.profile.engine, env=env_name)

    # Match run.py: simple profile/engine banner
    echo(f"Profile: {env_name} | Engine: {ctx.profile.engine}")

    # Reuse the same select token parsing as fft run (but purely static)
    select_tokens = _parse_select(select or [])

    # Run CI core checks on the loaded registry
    summary = run_ci_check(select=select_tokens)

    # Text summary (stdout)
    _print_text_summary(
        summary,
        project=str(ctx.project),
        select_tokens=select_tokens,
    )

    # Decide exit code: any error-level issue ⇒ non-zero exit for CI
    has_errors = any(issue.level == "error" for issue in summary.issues)

    clear_context()

    if has_errors:
        raise typer.Exit(1)

register 

register(app)

Register fft ci-check on the main Typer app.

Source code in src/fastflowtransform/cli/ci_cmd.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def register(app: typer.Typer) -> None:
    """
    Register `fft ci-check` on the main Typer app.
    """
    app.command(
        "ci-check",
        help=textwrap.dedent(
            """\
            Static CI check: parse models, validate DAG, and preview selection
            without touching the database.

            Examples:
              fft ci-check . --env dev
              fft ci-check . --env dev --select tag:finance
            """
        ),
    )(ci_check)