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

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.
    """

    wall_seconds: Mapped[float | None] = mapped_column(Float, nullable=True)
    """
    Wall clock time taken by the execution, in seconds.

    Measured by the worker as a monotonic delta around the diagnostic run.
    """

    cpu_seconds: Mapped[float | None] = mapped_column(Float, nullable=True)
    """
    CPU time consumed by the execution, in seconds.

    Sums user and system time for the worker process and its children.
    """

    peak_memory_bytes: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
    """
    Peak resident memory observed during the execution, in bytes.

    Stored as a big integer because a plain integer overflows at 4 GiB on Postgres,
    and executions routinely peak well above that.
    """

    memory_source: Mapped[str | None] = mapped_column(nullable=True)
    """
    Provenance of ``peak_memory_bytes``.

    One of ``cgroup``, ``proc_tree``, ``rusage`` or ``unavailable``.
    The sources measure different things and are not interchangeable.
    """

    memory_limit_bytes: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
    """
    Memory limit in force while the execution ran, in bytes.

    Read from the container cgroup at run time.
    A column rather than a context key because headroom is a comparison against
    ``peak_memory_bytes`` that has to happen in SQL.
    """

    cpu_limit: Mapped[float | None] = mapped_column(Float, nullable=True)
    """
    CPU cores available to the execution.

    Read from the container cgroup quota at run time.
    """

    resources_exclusive: Mapped[bool | None] = mapped_column(nullable=True)
    """
    Whether this execution had the worker's cgroup to itself.

    Declared by the executor, which is the only party that knows its own concurrency.
    A cgroup reading covers the whole container,
    so it is meaningless when a container runs several executions at once.
    Non-exclusive measurements are excluded from aggregation by default.
    """

    queue_seconds: Mapped[float | None] = mapped_column(Float, nullable=True)
    """
    Time between submission and the start of the execution, in seconds.

    Recorded by the submitting executor rather than the worker.
    """

    resource_context: Mapped[dict[str, Any] | None] = mapped_column(nullable=True)
    """
    Supporting detail for the resource measurements.

    Host, CPU count, sampler settings and any cross-checks the worker made.
    Both sampled peaks are kept here, under ``cgroup_peak_bytes`` and ``proc_tree_peak_bytes``,
    whichever of them ``peak_memory_bytes`` reports,
    so which one to trust stays a read-time decision.
    A large gap between the two says the cgroup was shared.
    Purely informational, so nothing here is a query predicate.
    """

    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 apply_resource_usage(
        self, usage: "ResourceUsage | None", *, queue_seconds: float | None = None
    ) -> None:
        """
        Copy a worker's resource measurements onto this row.

        Every value is coerced before any of them is assigned,
        so a value a column cannot store raises here rather than at flush,
        where the offending execution is no longer identifiable.

        Parameters
        ----------
        usage
            The measurements taken by the worker, or None when nothing was measured.
        queue_seconds
            Submit-to-start latency observed by the submitting executor,
            or None when it is unknown.
        """
        if queue_seconds is not None:
            self.queue_seconds = float(queue_seconds)

        if usage is None:
            return

        wall_seconds = float(usage.wall_seconds)
        cpu_seconds = None if usage.cpu_seconds is None else float(usage.cpu_seconds)
        peak_memory_bytes = None if usage.peak_memory_bytes is None else int(usage.peak_memory_bytes)
        memory_source = str(usage.memory_source)
        memory_limit_bytes = None if usage.memory_limit_bytes is None else int(usage.memory_limit_bytes)
        cpu_limit = None if usage.cpu_limit is None else float(usage.cpu_limit)
        exclusive = bool(usage.exclusive)
        context = dict(usage.context)

        self.wall_seconds = wall_seconds
        self.cpu_seconds = cpu_seconds
        self.peak_memory_bytes = peak_memory_bytes
        self.memory_source = memory_source
        self.memory_limit_bytes = memory_limit_bytes
        self.cpu_limit = cpu_limit
        self.resources_exclusive = exclusive
        self.resource_context = context

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

cpu_limit = mapped_column(Float, nullable=True) class-attribute instance-attribute

CPU cores available to the execution.

Read from the container cgroup quota at run time.

cpu_seconds = mapped_column(Float, nullable=True) class-attribute instance-attribute

CPU time consumed by the execution, in seconds.

Sums user and system time for the worker process and its children.

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

memory_limit_bytes = mapped_column(BigInteger, nullable=True) class-attribute instance-attribute

Memory limit in force while the execution ran, in bytes.

Read from the container cgroup at run time. A column rather than a context key because headroom is a comparison against peak_memory_bytes that has to happen in SQL.

memory_source = mapped_column(nullable=True) class-attribute instance-attribute

Provenance of peak_memory_bytes.

One of cgroup, proc_tree, rusage or unavailable. The sources measure different things and are not interchangeable.

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

peak_memory_bytes = mapped_column(BigInteger, nullable=True) class-attribute instance-attribute

Peak resident memory observed during the execution, in bytes.

Stored as a big integer because a plain integer overflows at 4 GiB on Postgres, and executions routinely peak well above that.

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.

queue_seconds = mapped_column(Float, nullable=True) class-attribute instance-attribute

Time between submission and the start of the execution, in seconds.

Recorded by the submitting executor rather than the worker.

resource_context = mapped_column(nullable=True) class-attribute instance-attribute

Supporting detail for the resource measurements.

Host, CPU count, sampler settings and any cross-checks the worker made. Both sampled peaks are kept here, under cgroup_peak_bytes and proc_tree_peak_bytes, whichever of them peak_memory_bytes reports, so which one to trust stays a read-time decision. A large gap between the two says the cgroup was shared. Purely informational, so nothing here is a query predicate.

resources_exclusive = mapped_column(nullable=True) class-attribute instance-attribute

Whether this execution had the worker's cgroup to itself.

Declared by the executor, which is the only party that knows its own concurrency. A cgroup reading covers the whole container, so it is meaningless when a container runs several executions at once. Non-exclusive measurements are excluded from aggregation by default.

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

wall_seconds = mapped_column(Float, nullable=True) class-attribute instance-attribute

Wall clock time taken by the execution, in seconds.

Measured by the worker as a monotonic delta around the diagnostic run.

apply_resource_usage(usage, *, queue_seconds=None)

Copy a worker's resource measurements onto this row.

Every value is coerced before any of them is assigned, so a value a column cannot store raises here rather than at flush, where the offending execution is no longer identifiable.

Parameters:

Name	Type	Description	Default
usage	ResourceUsage | None	The measurements taken by the worker, or None when nothing was measured.	required
queue_seconds	float | None	Submit-to-start latency observed by the submitting executor, or None when it is unknown.	None

Source code in packages/climate-ref/src/climate_ref/models/execution.py

def apply_resource_usage(
    self, usage: "ResourceUsage | None", *, queue_seconds: float | None = None
) -> None:
    """
    Copy a worker's resource measurements onto this row.

    Every value is coerced before any of them is assigned,
    so a value a column cannot store raises here rather than at flush,
    where the offending execution is no longer identifiable.

    Parameters
    ----------
    usage
        The measurements taken by the worker, or None when nothing was measured.
    queue_seconds
        Submit-to-start latency observed by the submitting executor,
        or None when it is unknown.
    """
    if queue_seconds is not None:
        self.queue_seconds = float(queue_seconds)

    if usage is None:
        return

    wall_seconds = float(usage.wall_seconds)
    cpu_seconds = None if usage.cpu_seconds is None else float(usage.cpu_seconds)
    peak_memory_bytes = None if usage.peak_memory_bytes is None else int(usage.peak_memory_bytes)
    memory_source = str(usage.memory_source)
    memory_limit_bytes = None if usage.memory_limit_bytes is None else int(usage.memory_limit_bytes)
    cpu_limit = None if usage.cpu_limit is None else float(usage.cpu_limit)
    exclusive = bool(usage.exclusive)
    context = dict(usage.context)

    self.wall_seconds = wall_seconds
    self.cpu_seconds = cpu_seconds
    self.peak_memory_bytes = peak_memory_bytes
    self.memory_source = memory_source
    self.memory_limit_bytes = memory_limit_bytes
    self.cpu_limit = cpu_limit
    self.resources_exclusive = exclusive
    self.resource_context = context

mark_failed()

Mark the diagnostic execution as unsuccessful

Source code in packages/climate-ref/src/climate_ref/models/execution.py

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

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

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

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

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

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

@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

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

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

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