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 
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
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
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 
278
279
280
281
282
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 
270
271
272
273
274
275
276
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 
260
261
262
263
264
265
266
267
268
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 
 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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
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, Execution.id"
    )

    def should_run(
        self,
        dataset_hash: str,
        rerun_failed: bool = False,
        stale_cutoff: "datetime.datetime | None" = None,
    ) -> 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)

        Parameters
        ----------
        dataset_hash
            Hash of the candidate datasets for this run.
        rerun_failed
            Re-run the group even if the last execution failed and the group is not dirty.
        stale_cutoff
            When provided,
            an in-progress execution created before this timestamp is treated as already failed.
            A real solve reaps such abandoned executions (via ``fail_stale_in_progress_executions``)
            before evaluating this method.
        """
        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

        treat_as_failed = (
            last_execution.successful is None
            and stale_cutoff is not None
            and last_execution.created_at < stale_cutoff
        )

        # 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 and not treat_as_failed:
            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 or treat_as_failed) 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, stale_cutoff=None) #

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)

Parameters:

Name Type Description Default
dataset_hash str

Hash of the candidate datasets for this run.

 required
rerun_failed bool

Re-run the group even if the last execution failed and the group is not dirty.

 False
stale_cutoff datetime | None

When provided, an in-progress execution created before this timestamp is treated as already failed. A real solve reaps such abandoned executions (via fail_stale_in_progress_executions) before evaluating this method.

 None
Source code in packages/climate-ref/src/climate_ref/models/execution.py 
 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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def should_run(
    self,
    dataset_hash: str,
    rerun_failed: bool = False,
    stale_cutoff: "datetime.datetime | None" = None,
) -> 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)

    Parameters
    ----------
    dataset_hash
        Hash of the candidate datasets for this run.
    rerun_failed
        Re-run the group even if the last execution failed and the group is not dirty.
    stale_cutoff
        When provided,
        an in-progress execution created before this timestamp is treated as already failed.
        A real solve reaps such abandoned executions (via ``fail_stale_in_progress_executions``)
        before evaluating this method.
    """
    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

    treat_as_failed = (
        last_execution.successful is None
        and stale_cutoff is not None
        and last_execution.created_at < stale_cutoff
    )

    # 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 and not treat_as_failed:
        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 or treat_as_failed) 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 
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
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
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 
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
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
@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 
285
286
287
288
289
290
291
292
293
294
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, among_executions=None) #

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
among_executions Sequence[Any] | None

Optional predicates on Execution applied before the latest-per-group ranking, so "latest" is chosen from that filtered population (see :func:_latest_execution_ids). Defaults to ranking over all executions.

 None

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 
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
def get_execution_group_and_latest(
    session: Session,
    among_executions: Sequence[Any] | None = None,
) -> 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.
    among_executions
        Optional predicates on ``Execution`` applied *before* the latest-per-group ranking,
        so "latest" is chosen from that filtered population (see :func:`_latest_execution_ids`).
        Defaults to ranking over all executions.

    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.
    """
    latest = _latest_execution_ids(session, among_executions)

    # Groups with no matching execution still appear (outer join, Execution=None).
    # Every known caller (`cli/executions.py::delete_groups`, `ExecutionsReader.groups()`,
    # `executor/reingest.py`) reads `eg.diagnostic.slug`/`eg.diagnostic.provider.slug` off the
    # returned groups, so eager-load both relationships here to avoid an N+1 lazy-load per row.
    query = (
        session.query(ExecutionGroup, Execution)
        .options(joinedload(ExecutionGroup.diagnostic).joinedload(Diagnostic.provider))
        .outerjoin(latest, latest.c.group_id == ExecutionGroup.id)
        .outerjoin(Execution, Execution.id == latest.c.execution_id)
    )

    return query  # type: ignore

get_execution_group_and_latest_filtered(session, diagnostic_filters=None, provider_filters=None, facet_filters=None, dirty=None, successful=None, latest_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.

Success can be filtered in two different ways: successful and latest_successful. successful=True keeps a group only if its newest run happened to succeed, whereas latest_successful=True changes which run is chosen as newest. The two compose but answer different questions.

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

Post-rank filter on the winning execution -- asks "is the latest execution 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.

 None
latest_successful bool | None

Pre-rank population filter -- asks "what is the latest successful execution?". If True, rank only over successful executions, so the returned execution is each group's latest successful run (if any). If False, rank only over unsuccessful / in-progress executions. If None (default), rank over all executions.

 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 
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
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,
    latest_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.

    Success can be filtered in two different ways: ``successful`` and ``latest_successful``.
    ``successful=True`` keeps a group only if its newest run happened to succeed,
    whereas ``latest_successful=True`` changes which run is chosen as newest.
    The two compose but answer different questions.

    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
        Post-rank filter on the *winning* execution -- asks "is the latest execution 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.
    latest_successful
        Pre-rank population filter -- asks "what is the latest *successful* execution?".
        If True, rank only over successful executions,
        so the returned execution is each group's latest successful run (if any).
        If False, rank only over unsuccessful / in-progress executions.
        If None (default), rank over all executions.
    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.
    """
    # Pre-rank population filter: restrict which executions "latest" is chosen from.
    among_executions = None if latest_successful is None else [_successful_predicate(latest_successful)]

    # Start with base query
    query = get_execution_group_and_latest(session, among_executions=among_executions)

    # 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:
        query = query.filter(_successful_predicate(successful))

    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()]