Skip to content

climate_ref.models.execution #

Execution #

Bases: CreatedUpdatedMixin, Base

Represents a single execution of a diagnostic

Each result is part of a group of executions that share similar input datasets.

An execution group might be run multiple times as new data becomes available, each run will create a Execution.

Source code in packages/climate-ref/src/climate_ref/models/execution.py 
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
185
186
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
class Execution(CreatedUpdatedMixin, Base):
    """
    Represents a single execution of a diagnostic

    Each result is part of a group of executions that share similar input datasets.

    An execution group might be run multiple times as new data becomes available,
    each run will create a `Execution`.
    """

    __tablename__ = "execution"

    id: Mapped[int] = mapped_column(primary_key=True)

    output_fragment: Mapped[str] = mapped_column()
    """
    Relative directory to store the output of the execution.

    During execution this directory is relative to the temporary directory.
    If the diagnostic execution is successful, the executions will be moved to the final output directory
    and the temporary directory will be cleaned up.
    This directory may contain multiple input and output files.
    """

    execution_group_id: Mapped[int] = mapped_column(
        ForeignKey(
            "execution_group.id",
            name="fk_execution_id",
        ),
        index=True,
    )
    """
    The execution group that this execution belongs to
    """

    dataset_hash: Mapped[str] = mapped_column(index=True)
    """
    Hash of the datasets used to calculate the diagnostic

    This is used to verify if an existing diagnostic execution has been run with the same datasets.
    """

    successful: Mapped[bool | None] = mapped_column(nullable=True, index=True)
    """
    Was the run successful
    """

    path: Mapped[str] = mapped_column(nullable=True)
    """
    Path to the output bundle

    Relative to the diagnostic execution result output directory
    """

    retracted: Mapped[bool] = mapped_column(default=False)
    """
    Whether the diagnostic execution result has been retracted or not

    This may happen if a dataset has been retracted, or if the diagnostic execution was incorrect.
    Rather than delete the values, they are marked as retracted.
    These data may still be visible in the UI, but should be marked as retracted.
    """

    provider_version: Mapped[str | None] = mapped_column(nullable=True)
    """
    Provider version recorded by the worker at run time.

    Snapshot of the worker-installed ``provider.version`` when the execution ran.
    Purely informational for audit; not used for validation or recomputation triggers.
    Rows that predate the column stay NULL.
    """

    execution_group: Mapped["ExecutionGroup"] = relationship(back_populates="executions")
    outputs: Mapped[list["ExecutionOutput"]] = relationship(back_populates="execution")
    values: Mapped[list["MetricValue"]] = relationship(back_populates="execution")

    datasets: Mapped[list[Dataset]] = relationship(secondary=execution_datasets)
    """
    The datasets used in this execution
    """

    def register_datasets(self, db: "Database", execution_dataset: "ExecutionDatasetCollection") -> None:
        """
        Register the datasets used in the diagnostic calculation with the execution
        """
        for _, dataset in execution_dataset.items():
            db.session.execute(
                execution_datasets.insert(),
                [{"execution_id": self.id, "dataset_id": idx} for idx in dataset.index],
            )

    def mark_successful(self, path: pathlib.Path | str) -> None:
        """
        Mark the diagnostic execution as successful
        """
        # TODO: this needs to accept both a diagnostic and output bundle
        self.successful = True
        self.path = str(path)

    def mark_failed(self) -> None:
        """
        Mark the diagnostic execution as unsuccessful
        """
        self.successful = False

dataset_hash = mapped_column(index=True) class-attribute instance-attribute #

Hash of the datasets used to calculate the diagnostic

This is used to verify if an existing diagnostic execution has been run with the same datasets.

datasets = relationship(secondary=execution_datasets) class-attribute instance-attribute #

The datasets used in this execution

execution_group_id = mapped_column(ForeignKey('execution_group.id', name='fk_execution_id'), index=True) class-attribute instance-attribute #

The execution group that this execution belongs to

output_fragment = mapped_column() class-attribute instance-attribute #

Relative directory to store the output of the execution.

During execution this directory is relative to the temporary directory. If the diagnostic execution is successful, the executions will be moved to the final output directory and the temporary directory will be cleaned up. This directory may contain multiple input and output files.

path = mapped_column(nullable=True) class-attribute instance-attribute #

Path to the output bundle

Relative to the diagnostic execution result output directory

provider_version = mapped_column(nullable=True) class-attribute instance-attribute #

Provider version recorded by the worker at run time.

Snapshot of the worker-installed provider.version when the execution ran. Purely informational for audit; not used for validation or recomputation triggers. Rows that predate the column stay NULL.

retracted = mapped_column(default=False) class-attribute instance-attribute #

Whether the diagnostic execution result has been retracted or not

This may happen if a dataset has been retracted, or if the diagnostic execution was incorrect. Rather than delete the values, they are marked as retracted. These data may still be visible in the UI, but should be marked as retracted.

successful = mapped_column(nullable=True, index=True) class-attribute instance-attribute #

Was the run successful

mark_failed() #

Mark the diagnostic execution as unsuccessful

Source code in packages/climate-ref/src/climate_ref/models/execution.py 
254
255
256
257
258
def mark_failed(self) -> None:
    """
    Mark the diagnostic execution as unsuccessful
    """
    self.successful = False

mark_successful(path) #

Mark the diagnostic execution as successful

Source code in packages/climate-ref/src/climate_ref/models/execution.py 
246
247
248
249
250
251
252
def mark_successful(self, path: pathlib.Path | str) -> None:
    """
    Mark the diagnostic execution as successful
    """
    # TODO: this needs to accept both a diagnostic and output bundle
    self.successful = True
    self.path = str(path)

register_datasets(db, execution_dataset) #

Register the datasets used in the diagnostic calculation with the execution

Source code in packages/climate-ref/src/climate_ref/models/execution.py 
236
237
238
239
240
241
242
243
244
def register_datasets(self, db: "Database", execution_dataset: "ExecutionDatasetCollection") -> None:
    """
    Register the datasets used in the diagnostic calculation with the execution
    """
    for _, dataset in execution_dataset.items():
        db.session.execute(
            execution_datasets.insert(),
            [{"execution_id": self.id, "dataset_id": idx} for idx in dataset.index],
        )

ExecutionGroup #

Bases: CreatedUpdatedMixin, Base

Represents a group of executions with a shared set of input datasets.

When solving, the ExecutionGroups are derived from the available datasets, the defined diagnostics and their data requirements. From the information in the group an execution can be triggered, which is an actual run of a diagnostic calculation with a specific set of input datasets.

When the ExecutionGroup is created, it is marked dirty, meaning there are no current executions available. When an Execution was run successfully for a ExecutionGroup, the dirty mark is removed. After ingesting new data and solving again and if new versions of the input datasets are available, the ExecutionGroup will be marked dirty again.

The diagnostic_id and key form a unique identifier for ExecutionGroups.

Source code in packages/climate-ref/src/climate_ref/models/execution.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 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
 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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
class ExecutionGroup(CreatedUpdatedMixin, Base):
    """
    Represents a group of executions with a shared set of input datasets.

    When solving, the `ExecutionGroup`s are derived from the available datasets,
    the defined diagnostics and their data requirements. From the information in the
    group an execution can be triggered, which is an actual run of a diagnostic calculation
    with a specific set of input datasets.

    When the `ExecutionGroup` is created, it is marked dirty, meaning there are no
    current executions available. When an Execution was run successfully for a
    ExecutionGroup, the dirty mark is removed. After ingesting new data and
    solving again and if new versions of the input datasets are available, the
    ExecutionGroup will be marked dirty again.

    The diagnostic_id and key form a unique identifier for `ExecutionGroup`s.
    """

    __tablename__ = "execution_group"
    __table_args__ = (UniqueConstraint("diagnostic_id", "key", "diagnostic_version", name="execution_ident"),)

    id: Mapped[int] = mapped_column(primary_key=True)

    diagnostic_id: Mapped[int] = mapped_column(ForeignKey("diagnostic.id"), index=True)
    """
    The diagnostic that this execution group belongs to
    """

    key: Mapped[str] = mapped_column(index=True)
    """
    Key for the datasets in this Execution group.
    """

    diagnostic_version: Mapped[int] = mapped_column(default=1, server_default="1")
    """
    Diagnostic version that produced this group.

    Read from the live ``Diagnostic.version`` class attribute at solve time.
    Combined with ``diagnostic_id`` and ``key`` to form the unique identifier,
    so v1 and v2 groups for the same key coexist as separate rows.
    """

    dirty: Mapped[bool] = mapped_column(default=False)
    """
    Whether the execution group should be rerun

    An execution group is dirty if the diagnostic or any of the input datasets has been
    updated since the last execution.
    """

    selectors: Mapped[dict[str, Any]] = mapped_column(default=dict)
    """
    Collection of selectors that define the group

    These selectors are the unique key, value pairs that were selected during the initial groupby
    operation.
    These are also used to define the dataset key.
    """

    diagnostic: Mapped["Diagnostic"] = relationship(back_populates="execution_groups")
    executions: Mapped[list["Execution"]] = relationship(
        back_populates="execution_group", order_by="Execution.created_at"
    )

    def should_run(self, dataset_hash: str, rerun_failed: bool = False) -> bool:
        """
        Check if the diagnostic execution group needs to be executed.

        The dirty flag is the primary signal for whether an execution group needs to be rerun.
        It is set when the group is created or when new data is available,
        and cleared when an execution completes (whether successful or not).
        Manual intervention (``flag-dirty``, ``fail-running``) can set it back to True.

        The execution group should be run if:

        * no executions have been performed ever
        * the dataset hash is different from the last run
        * the execution group is marked as dirty
        * ``rerun_failed=True`` is passed and the last execution failed

        The execution group should NOT be run if:

        * an execution with the same dataset hash is already in progress
        * the last execution failed and the group is not dirty
          (use ``rerun_failed=True`` or ``flag-dirty`` to retry)
        """
        if not self.executions:
            logger.debug(f"Execution group {self.diagnostic.slug}/{self.key} was never executed")
            return True

        last_execution = self.executions[-1]

        if last_execution.dataset_hash != dataset_hash:
            logger.debug(
                f"Execution group {self.diagnostic.slug}/{self.key} hash mismatch:"
                f" {last_execution.dataset_hash} != {dataset_hash}"
            )
            return True

        # Don't submit duplicate tasks for an execution that is already in progress
        # Stuck tasks can be cleaned up with the `fail-running` command
        if last_execution.successful is None:
            logger.debug(
                f"Execution group {self.diagnostic.slug}/{self.key} "
                f"already has an in-progress execution with hash {dataset_hash}"
            )
            return False

        # Dirty flag is the primary signal for rerunning existing jobs
        if self.dirty:
            logger.debug(f"Execution group {self.diagnostic.slug}/{self.key} is dirty")
            return True

        # Re-run all failed executions if explicitly requested
        if last_execution.successful is False and rerun_failed:
            logger.debug(
                f"Execution group {self.diagnostic.slug}/{self.key} "
                f"last execution failed, rerunning (rerun_failed=True)"
            )
            return True

        return False

diagnostic_id = mapped_column(ForeignKey('diagnostic.id'), index=True) class-attribute instance-attribute #

The diagnostic that this execution group belongs to

diagnostic_version = mapped_column(default=1, server_default='1') class-attribute instance-attribute #

Diagnostic version that produced this group.

Read from the live Diagnostic.version class attribute at solve time. Combined with diagnostic_id and key to form the unique identifier, so v1 and v2 groups for the same key coexist as separate rows.

dirty = mapped_column(default=False) class-attribute instance-attribute #

Whether the execution group should be rerun

An execution group is dirty if the diagnostic or any of the input datasets has been updated since the last execution.

key = mapped_column(index=True) class-attribute instance-attribute #

Key for the datasets in this Execution group.

selectors = mapped_column(default=dict) class-attribute instance-attribute #

Collection of selectors that define the group

These selectors are the unique key, value pairs that were selected during the initial groupby operation. These are also used to define the dataset key.

should_run(dataset_hash, rerun_failed=False) #

Check if the diagnostic execution group needs to be executed.

The dirty flag is the primary signal for whether an execution group needs to be rerun. It is set when the group is created or when new data is available, and cleared when an execution completes (whether successful or not). Manual intervention (flag-dirty, fail-running) can set it back to True.

The execution group should be run if:

  • no executions have been performed ever
  • the dataset hash is different from the last run
  • the execution group is marked as dirty
  • rerun_failed=True is passed and the last execution failed

The execution group should NOT be run if:

  • an execution with the same dataset hash is already in progress
  • the last execution failed and the group is not dirty (use rerun_failed=True or flag-dirty to retry)
Source code in packages/climate-ref/src/climate_ref/models/execution.py 
 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
141
142
143
144
def should_run(self, dataset_hash: str, rerun_failed: bool = False) -> bool:
    """
    Check if the diagnostic execution group needs to be executed.

    The dirty flag is the primary signal for whether an execution group needs to be rerun.
    It is set when the group is created or when new data is available,
    and cleared when an execution completes (whether successful or not).
    Manual intervention (``flag-dirty``, ``fail-running``) can set it back to True.

    The execution group should be run if:

    * no executions have been performed ever
    * the dataset hash is different from the last run
    * the execution group is marked as dirty
    * ``rerun_failed=True`` is passed and the last execution failed

    The execution group should NOT be run if:

    * an execution with the same dataset hash is already in progress
    * the last execution failed and the group is not dirty
      (use ``rerun_failed=True`` or ``flag-dirty`` to retry)
    """
    if not self.executions:
        logger.debug(f"Execution group {self.diagnostic.slug}/{self.key} was never executed")
        return True

    last_execution = self.executions[-1]

    if last_execution.dataset_hash != dataset_hash:
        logger.debug(
            f"Execution group {self.diagnostic.slug}/{self.key} hash mismatch:"
            f" {last_execution.dataset_hash} != {dataset_hash}"
        )
        return True

    # Don't submit duplicate tasks for an execution that is already in progress
    # Stuck tasks can be cleaned up with the `fail-running` command
    if last_execution.successful is None:
        logger.debug(
            f"Execution group {self.diagnostic.slug}/{self.key} "
            f"already has an in-progress execution with hash {dataset_hash}"
        )
        return False

    # Dirty flag is the primary signal for rerunning existing jobs
    if self.dirty:
        logger.debug(f"Execution group {self.diagnostic.slug}/{self.key} is dirty")
        return True

    # Re-run all failed executions if explicitly requested
    if last_execution.successful is False and rerun_failed:
        logger.debug(
            f"Execution group {self.diagnostic.slug}/{self.key} "
            f"last execution failed, rerunning (rerun_failed=True)"
        )
        return True

    return False

ExecutionOutput #

Bases: DimensionMixin, CreatedUpdatedMixin, Base

An output generated as part of an execution.

This output may be a plot, data file or HTML file. These outputs are defined in the CMEC output bundle.

Outputs can be tagged with dimensions from the controlled vocabulary to enable filtering and organization.

Source code in packages/climate-ref/src/climate_ref/models/execution.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
class ExecutionOutput(DimensionMixin, CreatedUpdatedMixin, Base):
    """
    An output generated as part of an execution.

    This output may be a plot, data file or HTML file.
    These outputs are defined in the CMEC output bundle.

    Outputs can be tagged with dimensions from the controlled vocabulary
    to enable filtering and organization.
    """

    __tablename__ = "execution_output"

    _cv_dimensions: ClassVar[list[str]] = []

    id: Mapped[int] = mapped_column(primary_key=True)

    execution_id: Mapped[int] = mapped_column(ForeignKey("execution.id"), index=True)

    output_type: Mapped[ResultOutputType] = mapped_column(index=True)
    """
    Type of the output

    This will determine how the output is displayed
    """

    filename: Mapped[str] = mapped_column(nullable=True)
    """
    Path to the output

    Relative to the diagnostic execution result output directory
    """

    short_name: Mapped[str] = mapped_column(nullable=True)
    """
    Short key of the output

    This is unique for a given result and output type
    """

    long_name: Mapped[str] = mapped_column(nullable=True)
    """
    Human readable name describing the plot
    """

    description: Mapped[str] = mapped_column(nullable=True)
    """
    Long description describing the plot
    """

    execution: Mapped["Execution"] = relationship(back_populates="outputs")

    @classmethod
    def build(  # noqa: PLR0913
        cls,
        *,
        execution_id: int,
        output_type: ResultOutputType,
        dimensions: dict[str, str],
        filename: str | None = None,
        short_name: str | None = None,
        long_name: str | None = None,
        description: str | None = None,
    ) -> "ExecutionOutput":
        """
        Build an ExecutionOutput from dimensions and metadata

        This is a helper method that validates the dimensions supplied.

        Parameters
        ----------
        execution_id
            Execution that created the output
        output_type
            Type of the output
        dimensions
            Dimensions that describe the output
        filename
            Path to the output
        short_name
            Short key of the output
        long_name
            Human readable name
        description
            Long description

        Raises
        ------
        KeyError
            If an unknown dimension was supplied.

            Dimensions must exist in the controlled vocabulary.

        Returns
        -------
            Newly created ExecutionOutput
        """
        for k in dimensions:
            if k not in cls._cv_dimensions:
                raise KeyError(f"Unknown dimension column '{k}'")

        return ExecutionOutput(
            execution_id=execution_id,
            output_type=output_type,
            filename=filename,
            short_name=short_name,
            long_name=long_name,
            description=description,
            **dimensions,
        )

description = mapped_column(nullable=True) class-attribute instance-attribute #

Long description describing the plot

filename = mapped_column(nullable=True) class-attribute instance-attribute #

Path to the output

Relative to the diagnostic execution result output directory

long_name = mapped_column(nullable=True) class-attribute instance-attribute #

Human readable name describing the plot

output_type = mapped_column(index=True) class-attribute instance-attribute #

Type of the output

This will determine how the output is displayed

short_name = mapped_column(nullable=True) class-attribute instance-attribute #

Short key of the output

This is unique for a given result and output type

build(*, execution_id, output_type, dimensions, filename=None, short_name=None, long_name=None, description=None) classmethod #

Build an ExecutionOutput from dimensions and metadata

This is a helper method that validates the dimensions supplied.

Parameters:

Name Type Description Default
execution_id int

Execution that created the output

 required
output_type ResultOutputType

Type of the output

 required
dimensions dict[str, str]

Dimensions that describe the output

 required
filename str | None

Path to the output

 None
short_name str | None

Short key of the output

 None
long_name str | None

Human readable name

 None
description str | None

Long description

 None

Raises:

Type Description
KeyError

If an unknown dimension was supplied.

Dimensions must exist in the controlled vocabulary.

Returns:

Type Description
Newly created ExecutionOutput
Source code in packages/climate-ref/src/climate_ref/models/execution.py 
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
@classmethod
def build(  # noqa: PLR0913
    cls,
    *,
    execution_id: int,
    output_type: ResultOutputType,
    dimensions: dict[str, str],
    filename: str | None = None,
    short_name: str | None = None,
    long_name: str | None = None,
    description: str | None = None,
) -> "ExecutionOutput":
    """
    Build an ExecutionOutput from dimensions and metadata

    This is a helper method that validates the dimensions supplied.

    Parameters
    ----------
    execution_id
        Execution that created the output
    output_type
        Type of the output
    dimensions
        Dimensions that describe the output
    filename
        Path to the output
    short_name
        Short key of the output
    long_name
        Human readable name
    description
        Long description

    Raises
    ------
    KeyError
        If an unknown dimension was supplied.

        Dimensions must exist in the controlled vocabulary.

    Returns
    -------
        Newly created ExecutionOutput
    """
    for k in dimensions:
        if k not in cls._cv_dimensions:
            raise KeyError(f"Unknown dimension column '{k}'")

    return ExecutionOutput(
        execution_id=execution_id,
        output_type=output_type,
        filename=filename,
        short_name=short_name,
        long_name=long_name,
        description=description,
        **dimensions,
    )

ResultOutputType #

Bases: Enum

Types of supported outputs

These map to the categories of output in the CMEC output bundle

Source code in packages/climate-ref/src/climate_ref/models/execution.py 
261
262
263
264
265
266
267
268
269
270
class ResultOutputType(enum.Enum):
    """
    Types of supported outputs

    These map to the categories of output in the CMEC output bundle
    """

    Plot = "plot"
    Data = "data"
    HTML = "html"

get_execution_group_and_latest(session) #

Query to get the most recent result for each execution group

Parameters:

Name Type Description Default
session Session

The database session to use for the query.

 required

Returns:

Type Description
Query to get the most recent result for each execution group.

The result is a tuple of the execution group and the most recent result, which can be None.
Source code in packages/climate-ref/src/climate_ref/models/execution.py 
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
def get_execution_group_and_latest(
    session: Session,
) -> RowReturningQuery[tuple[ExecutionGroup, Execution | None]]:
    """
    Query to get the most recent result for each execution group

    Parameters
    ----------
    session
        The database session to use for the query.

    Returns
    -------
        Query to get the most recent result for each execution group.
        The result is a tuple of the execution group and the most recent result,
        which can be None.
    """
    # Find the most recent result for each execution group
    # This uses an aggregate function because it is more efficient than order by
    subquery = (
        session.query(
            Execution.execution_group_id,
            func.max(Execution.created_at).label("latest_created_at"),
        )
        .group_by(Execution.execution_group_id)
        .subquery()
    )

    # Join the diagnostic execution with the latest result
    query = (
        session.query(ExecutionGroup, Execution)
        .outerjoin(subquery, ExecutionGroup.id == subquery.c.execution_group_id)
        .outerjoin(
            Execution,
            (Execution.execution_group_id == ExecutionGroup.id)
            & (Execution.created_at == subquery.c.latest_created_at),
        )
    )

    return query  # type: ignore

get_execution_group_and_latest_filtered(session, diagnostic_filters=None, provider_filters=None, facet_filters=None, dirty=None, successful=None, include_superseded=False) #

Query execution groups with filtering capabilities.

By default, returns only execution groups whose diagnostic_version matches the parent diagnostic's promoted_version so consumers see exactly one version's worth of results. Pass include_superseded=True to bypass the version filter and see the full history.

Parameters:

Name Type Description Default
session Session

Database session

 required
diagnostic_filters list[str] | None

List of diagnostic slug substrings (OR logic, case-insensitive)

 None
provider_filters list[str] | None

List of provider slug substrings (OR logic, case-insensitive)

 None
facet_filters dict[str, list[str]] | None

Dictionary mapping facet keys to lists of allowed values. Different keys are ANDed; multiple values for the same key are ORed.

 None
dirty bool | None

If True, only return dirty execution groups. If False, only return clean execution groups. If None, do not filter by dirty status.

 None
successful bool | None

If True, only return execution groups whose latest execution was successful. If False, only return execution groups whose latest execution was unsuccessful or has no executions. If None, do not filter by execution success.

 None
include_superseded bool

If True, include execution groups for diagnostic versions older than the currently promoted version. If False (default), join Diagnostic and filter to ExecutionGroup.diagnostic_version == Diagnostic.promoted_version. Set this for recovery / audit callers that need the full version history (e.g. executor/reingest.py).

 False

Returns:

Type Description
Query returning tuples of (ExecutionGroup, latest Execution or None)
Notes
  • Diagnostic and provider filters use substring matching (case-insensitive)
  • Multiple values within same filter type use OR logic
  • Different filter types use AND logic
  • Facet filters can either be key=value (searches all dataset types) or dataset_type.key=value (searches specific dataset type)
  • This helper is the only sanctioned path for new callers that should respect the promoted-version filter. The one acknowledged exception is the cli/executions.py::stats aggregation, which inlines .join(Diagnostic).filter(ExecutionGroup.diagnostic_version == Diagnostic.promoted_version) because it returns aggregate rows rather than a list of tuples and so cannot reuse this helper. Operational queries that must remain version-agnostic (e.g. mark_failed_running in the same module) intentionally do not use this helper at all.
Source code in packages/climate-ref/src/climate_ref/models/execution.py 
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
def get_execution_group_and_latest_filtered(  # noqa: PLR0913
    session: Session,
    diagnostic_filters: list[str] | None = None,
    provider_filters: list[str] | None = None,
    facet_filters: dict[str, list[str]] | None = None,
    dirty: bool | None = None,
    successful: bool | None = None,
    include_superseded: bool = False,
) -> list[tuple[ExecutionGroup, Execution | None]]:
    """
    Query execution groups with filtering capabilities.

    By default, returns only execution groups whose ``diagnostic_version`` matches
    the parent diagnostic's ``promoted_version`` so consumers see exactly one
    version's worth of results.
    Pass ``include_superseded=True`` to bypass the version filter and see the full history.

    Parameters
    ----------
    session
        Database session
    diagnostic_filters
        List of diagnostic slug substrings (OR logic, case-insensitive)
    provider_filters
        List of provider slug substrings (OR logic, case-insensitive)
    facet_filters
        Dictionary mapping facet keys to lists of allowed values.
        Different keys are ANDed; multiple values for the same key are ORed.
    dirty
        If True, only return dirty execution groups.
        If False, only return clean execution groups.
        If None, do not filter by dirty status.
    successful
        If True, only return execution groups whose latest execution was successful.
        If False, only return execution groups whose latest execution was unsuccessful or has no executions.
        If None, do not filter by execution success.
    include_superseded
        If True, include execution groups for diagnostic versions older than the
        currently promoted version.
        If False (default), join ``Diagnostic`` and filter to ``ExecutionGroup.diagnostic_version
        == Diagnostic.promoted_version``.
        Set this for recovery / audit callers that need the full version history
        (e.g. ``executor/reingest.py``).

    Returns
    -------
        Query returning tuples of (ExecutionGroup, latest Execution or None)

    Notes
    -----
    - Diagnostic and provider filters use substring matching (case-insensitive)
    - Multiple values within same filter type use OR logic
    - Different filter types use AND logic
    - Facet filters can either be key=value (searches all dataset types)
      or dataset_type.key=value (searches specific dataset type)
    - This helper is the only sanctioned path for new callers that should respect the promoted-version filter.
      The one acknowledged exception is the ``cli/executions.py::stats`` aggregation,
      which inlines
      ``.join(Diagnostic).filter(ExecutionGroup.diagnostic_version == Diagnostic.promoted_version)``
      because it returns aggregate rows rather than a list of tuples and so cannot reuse this helper.
      Operational queries that must remain version-agnostic
      (e.g. ``mark_failed_running`` in the same module) intentionally do not use this helper at all.
    """
    # Start with base query
    query = get_execution_group_and_latest(session)

    # Join Diagnostic when needed for filtering (by name or by promoted version).
    needs_diagnostic_join = bool(diagnostic_filters or provider_filters) or not include_superseded
    if needs_diagnostic_join:
        query = query.join(Diagnostic, ExecutionGroup.diagnostic_id == Diagnostic.id)

    if not include_superseded:
        query = query.filter(ExecutionGroup.diagnostic_version == Diagnostic.promoted_version)

    # Apply diagnostic filter (OR logic for multiple values)
    if diagnostic_filters:
        diagnostic_conditions = [
            Diagnostic.slug.ilike(f"%{filter_value.lower()}%") for filter_value in diagnostic_filters
        ]
        query = query.filter(or_(*diagnostic_conditions))

    # Apply provider filter (OR logic for multiple values)
    if provider_filters:
        # Need to join through Diagnostic to Provider
        query = query.join(Provider, Diagnostic.provider_id == Provider.id)

        provider_conditions = [
            Provider.slug.ilike(f"%{filter_value.lower()}%") for filter_value in provider_filters
        ]
        query = query.filter(or_(*provider_conditions))

    if successful is not None:
        if successful:
            query = query.filter(Execution.successful.is_(True))
        else:
            query = query.filter(or_(Execution.successful.is_(False), Execution.successful.is_(None)))

    if dirty is not None:
        if dirty:
            query = query.filter(ExecutionGroup.dirty.is_(True))
        else:
            query = query.filter(or_(ExecutionGroup.dirty.is_(False), ExecutionGroup.dirty.is_(None)))

    if facet_filters:
        # Load all results into memory for Python-based filtering
        # TODO: Update once we have normalised the selector
        results = [r._tuple() for r in query.all()]
        return _filter_executions_by_facets(results, facet_filters)
    else:
        return [r._tuple() for r in query.all()]