Skip to content

symfonic.diagnostics.runner

runner

Audit orchestrator for symfonic doctor (v7.16, item 5).

Imports every registered check coroutine from :mod:symfonic.diagnostics.checks, invokes them in a stable order, and aggregates the results into a single :class:AuditReport.

Network-sensitive checks (currently only check_postgres_dsn) run last so the faster pure-introspection checks are not blocked by a slow probe. When no_network=True (the default in CI), the check_postgres_dsn short-circuits and emits an INFO that the probe was skipped.

CheckFn module-attribute 

CheckFn = Callable[
    ["SymfonicAgent"], Awaitable[list[CheckResult]]
]

Type alias for the uniform check coroutine signature.

AuditRunner 

AuditRunner(*, no_network: bool = False)

Invoke every registered check against an agent and aggregate results.

The runner is constructed with the audit configuration (currently just no_network). run() is the async entry point; :func:audit_hms and :func:audit_hms_sync in the package __init__ are the public wrappers.

Source code in src/symfonic/diagnostics/runner.py 
63
64
def __init__(self, *, no_network: bool = False) -> None:
    self._no_network = no_network

run async 

run(agent: SymfonicAgent) -> AuditReport

Invoke every check; return an :class:AuditReport.

Source code in src/symfonic/diagnostics/runner.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
async def run(self, agent: SymfonicAgent) -> AuditReport:
    """Invoke every check; return an :class:`AuditReport`."""
    results: list[CheckResult] = []
    for fn in _LOCAL_CHECKS:
        findings = await fn(agent)
        results.extend(findings)

    if not self._no_network:
        for fn in _NETWORK_CHECKS:
            findings = await fn(agent)
            results.extend(findings)
    else:
        results.append(
            CheckResult(
                name="postgres_dsn.skipped",
                severity=Severity.INFO,
                message=(
                    "POSTGRES_DSN probe skipped (CI=true or --no-network)."
                ),
                fix_hint=None,
            ),
        )
    return AuditReport(results=results)

resolve_no_network 

resolve_no_network(
    *, explicit_flag: bool | None = None
) -> bool

Return whether network checks should be skipped.

Precedence:

  • If explicit_flag is not None, honour it verbatim.
  • Otherwise, check CI=true in the environment.

Most CI runners (GitHub Actions, GitLab, CircleCI, Buildkite) export CI=true; this gives the runner a sane default that doesn't require adopters to wire --no-network into every job config.

Source code in src/symfonic/diagnostics/runner.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def resolve_no_network(*, explicit_flag: bool | None = None) -> bool:
    """Return whether network checks should be skipped.

    Precedence:

    * If ``explicit_flag`` is not ``None``, honour it verbatim.
    * Otherwise, check ``CI=true`` in the environment.

    Most CI runners (GitHub Actions, GitLab, CircleCI, Buildkite) export
    ``CI=true``; this gives the runner a sane default that doesn't
    require adopters to wire ``--no-network`` into every job config.
    """
    if explicit_flag is not None:
        return explicit_flag
    return os.environ.get("CI", "").strip().lower() == "true"