Skip to content

climate_ref.results.diagnostics #

Typed, detached read surface for diagnostic metadata.

DiagnosticsReader is reached via Reader.diagnostics. It executes the select_diagnostics query (joining Diagnostic -> Provider and counting execution groups per diagnostic) and maps the rows into detached DTOs that outlive the session.

DiagnosticCollection #

An immutable page of diagnostics plus collection-level metadata.

total_count is the number of diagnostics matching the filter before pagination, so a caller can tell there are more rows than the returned page. offset/limit echo back the pagination applied to produce items (limit is None when the whole result was returned).

Source code in packages/climate-ref/src/climate_ref/results/diagnostics.py 
 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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@attrs.frozen(kw_only=True)
class DiagnosticCollection:
    """
    An immutable page of diagnostics plus collection-level metadata.

    ``total_count`` is the number of diagnostics matching the filter *before* pagination, so a
    caller can tell there are more rows than the returned page. ``offset``/``limit`` echo back the
    pagination applied to produce ``items`` (``limit`` is ``None`` when the whole result was
    returned).
    """

    items: tuple[DiagnosticView, ...]
    """The diagnostics on this page."""

    total_count: int
    """Total diagnostics matching the filter before ``offset``/``limit``."""

    offset: int
    """Rows skipped before this page."""

    limit: int | None
    """Page size requested, or ``None`` when the whole result was returned."""

    def __iter__(self) -> Iterator[DiagnosticView]:
        return iter(self.items)

    def __len__(self) -> int:
        return len(self.items)

    def to_pandas(self) -> pd.DataFrame:
        """
        DataFrame mirroring the ``diagnostics list`` CLI columns.

        Columns are emitted explicitly (``provider, diagnostic, name, promoted_version,
        execution_group_count, successful, inflight, total``) even when the collection is empty, so
        callers can select columns / build an empty table without special-casing.
        """
        columns = [
            "provider",
            "diagnostic",
            "name",
            "promoted_version",
            "execution_group_count",
            "successful",
            "inflight",
            "total",
        ]
        records = [
            {
                "provider": d.provider_slug,
                "diagnostic": d.slug,
                "name": d.name,
                "promoted_version": d.promoted_version,
                "execution_group_count": d.execution_group_count,
                "successful": d.successful,
                "inflight": d.inflight,
                "total": d.total,
            }
            for d in self.items
        ]
        return pd.DataFrame.from_records(records, columns=columns)

items instance-attribute #

The diagnostics on this page.

limit instance-attribute #

Page size requested, or None when the whole result was returned.

offset instance-attribute #

Rows skipped before this page.

total_count instance-attribute #

Total diagnostics matching the filter before offset/limit.

to_pandas() #

DataFrame mirroring the diagnostics list CLI columns.

Columns are emitted explicitly (provider, diagnostic, name, promoted_version, execution_group_count, successful, inflight, total) even when the collection is empty, so callers can select columns / build an empty table without special-casing.

Source code in packages/climate-ref/src/climate_ref/results/diagnostics.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def to_pandas(self) -> pd.DataFrame:
    """
    DataFrame mirroring the ``diagnostics list`` CLI columns.

    Columns are emitted explicitly (``provider, diagnostic, name, promoted_version,
    execution_group_count, successful, inflight, total``) even when the collection is empty, so
    callers can select columns / build an empty table without special-casing.
    """
    columns = [
        "provider",
        "diagnostic",
        "name",
        "promoted_version",
        "execution_group_count",
        "successful",
        "inflight",
        "total",
    ]
    records = [
        {
            "provider": d.provider_slug,
            "diagnostic": d.slug,
            "name": d.name,
            "promoted_version": d.promoted_version,
            "execution_group_count": d.execution_group_count,
            "successful": d.successful,
            "inflight": d.inflight,
            "total": d.total,
        }
        for d in self.items
    ]
    return pd.DataFrame.from_records(records, columns=columns)

DiagnosticFilter #

Declarative filter over diagnostics.

Every field is optional; None means "do not constrain on this axis". diagnostic_contains/provider_contains are case-insensitive substring matches (OR-combined within each field), matching the semantics used by ExecutionGroupFilter.

Source code in packages/climate-ref/src/climate_ref/results/diagnostics.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@attrs.frozen(kw_only=True)
class DiagnosticFilter:
    """
    Declarative filter over diagnostics.

    Every field is optional; ``None`` means "do not constrain on this axis".
    ``diagnostic_contains``/``provider_contains`` are case-insensitive substring matches
    (OR-combined within each field), matching the semantics used by
    [ExecutionGroupFilter][climate_ref.results.executions.ExecutionGroupFilter].
    """

    provider_contains: tuple[str, ...] | None = attrs.field(default=None, converter=_as_str_tuple)
    """Case-insensitive substring matches on provider slug (OR-combined)."""

    diagnostic_contains: tuple[str, ...] | None = attrs.field(default=None, converter=_as_str_tuple)
    """Case-insensitive substring matches on diagnostic slug (OR-combined)."""

diagnostic_contains = attrs.field(default=None, converter=_as_str_tuple) class-attribute instance-attribute #

Case-insensitive substring matches on diagnostic slug (OR-combined).

provider_contains = attrs.field(default=None, converter=_as_str_tuple) class-attribute instance-attribute #

Case-insensitive substring matches on provider slug (OR-combined).

DiagnosticView #

A single diagnostic with information about the currently promoted version and execution-group counts.

execution_group_count counts execution groups across every diagnostic version, while successful/inflight/total are scoped to the currently promoted_version.

Source code in packages/climate-ref/src/climate_ref/results/diagnostics.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
@attrs.frozen(kw_only=True)
class DiagnosticView:
    """
    A single diagnostic with information about the currently promoted version and execution-group counts.

    ``execution_group_count`` counts execution groups across every diagnostic version,
    while ``successful``/``inflight``/``total`` are scoped to the currently ``promoted_version``.
    """

    provider_slug: str
    """Owning provider's slug."""

    slug: str
    """The diagnostic's own slug, unique within its provider."""

    name: str
    """Human-readable diagnostic name."""

    promoted_version: int
    """Diagnostic version currently promoted for production."""

    execution_group_count: int
    """Execution groups for the diagnostic across every version."""

    successful: int
    """Promoted-version execution groups whose latest execution succeeded."""

    inflight: int
    """Promoted-version execution groups whose latest execution is still running (outcome not yet known)."""

    total: int
    """Execution groups at the promoted version."""

execution_group_count instance-attribute #

Execution groups for the diagnostic across every version.

inflight instance-attribute #

Promoted-version execution groups whose latest execution is still running (outcome not yet known).

name instance-attribute #

Human-readable diagnostic name.

promoted_version instance-attribute #

Diagnostic version currently promoted for production.

provider_slug instance-attribute #

Owning provider's slug.

slug instance-attribute #

The diagnostic's own slug, unique within its provider.

successful instance-attribute #

Promoted-version execution groups whose latest execution succeeded.

total instance-attribute #

Execution groups at the promoted version.

DiagnosticsReader #

Diagnostic read domain.

Constructed from a Database, which owns the session. All read methods return detached DTOs that outlive the session.

Source code in packages/climate-ref/src/climate_ref/results/diagnostics.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
class DiagnosticsReader:
    """
    Diagnostic read domain.

    Constructed from a [Database][climate_ref.database.Database], which owns the session.
    All read methods return detached DTOs that outlive the session.
    """

    def __init__(self, database: Database) -> None:
        self._db = database

    @property
    def session(self) -> Session:
        """The underlying database session."""
        return self._db.session

    def _to_view(self, row: Any, stats_by_key: Mapping[tuple[str, str], ExecutionStats]) -> DiagnosticView:
        stat = stats_by_key.get((row.provider_slug, row.slug))
        return DiagnosticView(
            provider_slug=row.provider_slug,
            slug=row.slug,
            name=row.name,
            promoted_version=row.promoted_version,
            execution_group_count=row.execution_group_count,
            successful=stat.successful if stat is not None else 0,
            inflight=stat.running if stat is not None else 0,
            total=stat.total if stat is not None else 0,
        )

    def list(
        self,
        filters: DiagnosticFilter | None = None,
        *,
        offset: int = 0,
        limit: int | None = None,
    ) -> DiagnosticCollection:
        """
        Query diagnostics, with their provider, execution-group count, and promoted-version stats.

        Pagination is applied in SQL (``offset``/``limit``), with ``total_count`` computed from a
        separate unpaged count query over the same filtered statement.
        Promoted-version status counts (``successful``/``inflight``/``total``) are merged in from
        [stats][climate_ref.results.diagnostics.DiagnosticsReader.stats],
        and keyed by ``(provider, diagnostic)`` so status-count logic stays defined once.

        Diagnostics with no execution groups at the promoted version get zeros.
        """
        filters = filters or DiagnosticFilter()

        base_stmt = select_diagnostics(filters)
        count_stmt = select(func.count()).select_from(base_stmt.subquery())
        total_count = self.session.execute(count_stmt).scalar_one()

        stmt = base_stmt
        if limit is not None:
            stmt = stmt.offset(offset).limit(limit)
        elif offset:
            stmt = stmt.offset(offset)

        rows = self.session.execute(stmt).all()

        stats = self.stats(
            provider_contains=filters.provider_contains,
            diagnostic_contains=filters.diagnostic_contains,
        )
        stats_by_key = {(s.provider, s.diagnostic): s for s in stats}
        items = tuple(self._to_view(row, stats_by_key) for row in rows)

        return DiagnosticCollection(items=items, total_count=total_count, offset=offset, limit=limit)

    def stats(
        self,
        *,
        diagnostic_contains: Sequence[str] | None = None,
        provider_contains: Sequence[str] | None = None,
    ) -> tuple[ExecutionStats, ...]:
        """
        Per-(provider, diagnostic) execution status counts.

        Reuses [select_execution_statistics][climate_ref.results.executions.select_execution_statistics]
        rather than duplicating the aggregate SQL, so status-count logic stays defined once.
        This has the same signature and behaviour as
        [ExecutionsReader.statistics][climate_ref.results.executions.ExecutionsReader.statistics],
        just reachable under the diagnostics domain.
        """
        stmt = select_execution_statistics(
            diagnostic_contains=diagnostic_contains, provider_contains=provider_contains
        )
        return rows_to_execution_stats(self.session.execute(stmt).all())

session property #

The underlying database session.

list(filters=None, *, offset=0, limit=None) #

Query diagnostics, with their provider, execution-group count, and promoted-version stats.

Pagination is applied in SQL (offset/limit), with total_count computed from a separate unpaged count query over the same filtered statement. Promoted-version status counts (successful/inflight/total) are merged in from stats, and keyed by (provider, diagnostic) so status-count logic stays defined once.

Diagnostics with no execution groups at the promoted version get zeros.

Source code in packages/climate-ref/src/climate_ref/results/diagnostics.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def list(
    self,
    filters: DiagnosticFilter | None = None,
    *,
    offset: int = 0,
    limit: int | None = None,
) -> DiagnosticCollection:
    """
    Query diagnostics, with their provider, execution-group count, and promoted-version stats.

    Pagination is applied in SQL (``offset``/``limit``), with ``total_count`` computed from a
    separate unpaged count query over the same filtered statement.
    Promoted-version status counts (``successful``/``inflight``/``total``) are merged in from
    [stats][climate_ref.results.diagnostics.DiagnosticsReader.stats],
    and keyed by ``(provider, diagnostic)`` so status-count logic stays defined once.

    Diagnostics with no execution groups at the promoted version get zeros.
    """
    filters = filters or DiagnosticFilter()

    base_stmt = select_diagnostics(filters)
    count_stmt = select(func.count()).select_from(base_stmt.subquery())
    total_count = self.session.execute(count_stmt).scalar_one()

    stmt = base_stmt
    if limit is not None:
        stmt = stmt.offset(offset).limit(limit)
    elif offset:
        stmt = stmt.offset(offset)

    rows = self.session.execute(stmt).all()

    stats = self.stats(
        provider_contains=filters.provider_contains,
        diagnostic_contains=filters.diagnostic_contains,
    )
    stats_by_key = {(s.provider, s.diagnostic): s for s in stats}
    items = tuple(self._to_view(row, stats_by_key) for row in rows)

    return DiagnosticCollection(items=items, total_count=total_count, offset=offset, limit=limit)

stats(*, diagnostic_contains=None, provider_contains=None) #

Per-(provider, diagnostic) execution status counts.

Reuses select_execution_statistics rather than duplicating the aggregate SQL, so status-count logic stays defined once. This has the same signature and behaviour as ExecutionsReader.statistics, just reachable under the diagnostics domain.

Source code in packages/climate-ref/src/climate_ref/results/diagnostics.py 
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
def stats(
    self,
    *,
    diagnostic_contains: Sequence[str] | None = None,
    provider_contains: Sequence[str] | None = None,
) -> tuple[ExecutionStats, ...]:
    """
    Per-(provider, diagnostic) execution status counts.

    Reuses [select_execution_statistics][climate_ref.results.executions.select_execution_statistics]
    rather than duplicating the aggregate SQL, so status-count logic stays defined once.
    This has the same signature and behaviour as
    [ExecutionsReader.statistics][climate_ref.results.executions.ExecutionsReader.statistics],
    just reachable under the diagnostics domain.
    """
    stmt = select_execution_statistics(
        diagnostic_contains=diagnostic_contains, provider_contains=provider_contains
    )
    return rows_to_execution_stats(self.session.execute(stmt).all())

select_diagnostics(filters=None) #

Build the Select for diagnostics joined to their provider, with an execution-group count.

execution_group_count counts every ExecutionGroup row for the diagnostic (all versions, not scoped to promoted_version), so it reflects the diagnostic's full execution history. Ordered by (Provider.slug, Diagnostic.slug, Diagnostic.id). The diagnostic primary key is the final tiebreak so SQL pagination is deterministic across pages.

Source code in packages/climate-ref/src/climate_ref/results/diagnostics.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def select_diagnostics(filters: DiagnosticFilter | None = None) -> Select[Any]:
    """
    Build the ``Select`` for diagnostics joined to their provider, with an execution-group count.

    ``execution_group_count`` counts every ``ExecutionGroup`` row for the diagnostic
    (all versions, not scoped to ``promoted_version``),
    so it reflects the diagnostic's full execution history.
    Ordered by ``(Provider.slug, Diagnostic.slug, Diagnostic.id)``.
    The diagnostic primary key is the final tiebreak so SQL pagination is deterministic across pages.
    """
    filters = filters or DiagnosticFilter()

    group_count_subquery = (
        select(
            ExecutionGroup.diagnostic_id,
            func.count(ExecutionGroup.id).label("execution_group_count"),
        )
        .group_by(ExecutionGroup.diagnostic_id)
        .subquery()
    )

    stmt = (
        select(
            Provider.slug.label("provider_slug"),
            Diagnostic.slug.label("slug"),
            Diagnostic.name.label("name"),
            Diagnostic.promoted_version.label("promoted_version"),
            func.coalesce(group_count_subquery.c.execution_group_count, 0).label("execution_group_count"),
        )
        .join(Provider, Diagnostic.provider_id == Provider.id)
        .outerjoin(group_count_subquery, Diagnostic.id == group_count_subquery.c.diagnostic_id)
        .order_by(Provider.slug, Diagnostic.slug, Diagnostic.id)
    )

    if filters.provider_contains:
        stmt = stmt.where(or_(*(Provider.slug.ilike(f"%{s.lower()}%") for s in filters.provider_contains)))
    if filters.diagnostic_contains:
        stmt = stmt.where(
            or_(*(Diagnostic.slug.ilike(f"%{s.lower()}%") for s in filters.diagnostic_contains))
        )

    return stmt