from __future__ import annotations
import copy
import enum
import threading
import weakref
from abc import ABC, abstractmethod
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from typing import Any, ClassVar, NamedTuple, cast
import apsw
from cerberus import Validator
from gain import logging
from gain.task_graph.graph import TaskDesc
from gain.templates import get_template
from gain.templates.breadcrumb import Crumb, page_breadcrumb
from gain.templates.static_assets import climb_to_root
from gain.utils.helpers import convert_size
from .dvc import is_dvc_sidecar
from .repository import (
GR_INDEX_FILE_NAME,
GR_INDEX_NON_LABEL_COLUMNS,
GR_INDEX_RESOURCE_FIELDS,
GR_STATISTICS_INDEX_FILE_NAME,
INDEX_COLUMN_PATTERN,
INDEX_COLUMN_RE,
GenomicResource,
GenomicResourceRepo,
_description_in,
_summary_in,
)
from .resource_errors import MalformedResourceError
from .resource_query import label_alternatives
from .statistics.base_statistic import StoredStatistic
logger = logging.getLogger(__name__)
ONE_REGION_PER_CONTIG = 3_000_000_000
"""A statistics region size larger than any contig, so every contig is one
region.
``split_into_regions`` clamps a region size to the contig length, so any
size at least as long as the longest contig builds exactly one region per
contig. This is not the ``0`` sentinel, which means "do not split": a
genomic score then scans the whole resource in one task, with no regions.
"""
DEFAULT_STATISTICS_REGION_SIZE = ONE_REGION_PER_CONTIG
"""The region size a statistics build uses when its caller names none: one
region per contig (gain#357).
A smaller size balances a single large resource across more workers; keep
it at or below about genome / (4 x workers).
"""
# Names FTS5 will not accept as a column of the index table: it reserves
# `rank` and `rowid`, and every FTS5 table has a hidden column named after
# the table itself -- which for the repository index is `contents`, the
# table _create_contents_db builds. Compared case-insensitively, as SQLite
# compares identifiers.
_INDEX_RESERVED_COLUMNS = frozenset({"rank", "rowid", "contents"})
# The index columns that describe the resource rather than one of its
# labels, lowercased for the case-insensitive comparison SQLite makes.
# A label key may never be one of these, whatever the resource's own
# implementation contributes -- see `collect_index_info` (gain#542).
_RESERVED_FIELD_NAMES = frozenset(
name.lower() for name in GR_INDEX_NON_LABEL_COLUMNS
)
[docs]
def get_base_resource_schema() -> dict[str, Any]:
return {
"type": {"type": "string"},
"meta": {
"type": "dict",
"allow_unknown": True,
"schema": {
"description": {"type": "string"},
# The keys of `labels` are constrained -- each one becomes a
# column of the repository's search index -- but deliberately
# NOT here: this schema is run by every implementation that
# validates, and for scores it runs inside
# ScoreResource.__init__, on the annotation path. A key that
# cannot name an index column would then make an otherwise
# sound resource unusable for annotation too. The rule is
# enforced where it bites, in collect_index_info() (gain#464).
"labels": {"type": "dict", "nullable": True},
},
},
}
[docs]
def get_required_filename_schema() -> dict[str, Any]:
"""The required ``filename`` config-schema fragment.
Spliced into a schema -- or a nested block of one -- wherever the loader
reads ``filename`` unconditionally. Not part of
``get_base_resource_schema()``, since not every implementation has a
file. Built fresh on every call: cerberus rewrites nested rule dicts
in place, on ``Validator`` construction and on every ``validate()``.
"""
return {"filename": {"type": "string", "required": True}}
def _index_column_problem(column: object, taken: set[str]) -> str | None:
"""Say why ``column`` cannot name a column of the FTS index, or None."""
if not isinstance(column, str):
# YAML mapping keys need not be strings: `2024: release` parses to
# an int key, `true:` to a bool, `null:` to None. Caught here so
# that the curator is told the rule -- letting one of these reach
# the regex raises a TypeError, which the index build can only
# report as an internal error (gain#464, gain#364).
return f"is not a string but {type(column).__name__}"
if not INDEX_COLUMN_RE.match(column):
return "is not a valid SQL identifier"
if column.upper() in apsw.keywords:
# Refusing every one of `apsw.keywords` is deliberately stricter
# than SQLite is. Measured against SQLite 3.53: FTS5 parses its own
# argument list, so the CREATE VIRTUAL TABLE accepts all 147 of
# them; it is the INSERT's column list, parsed by SQLite proper,
# that refuses the 58 reserved ones (`order`, `select`, `from`, ...)
# while the other 89 (`key`, `match`, `filter`, ...) work. Which
# keyword falls in which half is a property of the SQLite build and
# version, not of this code, so the whole list goes: the cost is a
# handful of label keys nobody uses -- no key in the live GRR is a
# keyword -- and the alternative is a rule that shifts under a
# dependency bump, in the direction of an unbuildable index.
return "is an SQL keyword"
if column.lower() in _INDEX_RESERVED_COLUMNS:
return "is a name FTS5 reserves"
if column.lower() in taken:
if column.lower() in _RESERVED_FIELD_NAMES:
# The name is one the index keeps for a resource field, and the
# curator's own resource may well have no such field -- a genome
# carrying a label named `score_ids` collides with a column only
# score resources fill. "Repeats a field" would leave them
# looking for a field that is not there, so name them (gain#542).
return (
"is a name the index reserves for a resource field "
f"({', '.join(sorted(_RESERVED_FIELD_NAMES))})"
)
return "repeats a field the index already has"
return None
[docs]
def validate_index_columns(
resource_id: str, columns: Sequence[str],
) -> None:
"""Check that ``columns`` can name the columns of the FTS index.
Every column name is interpolated into the ``CREATE VIRTUAL TABLE`` and
``INSERT`` statements that build the repository index, so a name that is
not a bare identifier -- and not one SQL or FTS5 has already taken -- is
at best unbuildable and at worst an injection (gain#464). Repeats are
rejected too: the index keeps one column per name. Where the repeated
name is a field of the resource itself, the label silently replaces that
field's value, which then cannot be found by it; where it is a name the
index reserves for a field some *other* implementation contributes, the
label lands in a column that means something else for every resource
that does contribute it (gain#542). Both are reported against the
offending column, the second naming the reserved set, since the curator
has no such field of their own to look at.
Raises ``ValueError`` naming the resource and the offending column. The
caller is the per-resource handler of the index build, so an offending
resource is skipped and reported by id instead of taking the whole
repository's index down with it.
The columns are typed as strings but come from a YAML mapping's keys,
which need not be -- a column that is not a string is refused like any
other bad name rather than raising out of the check.
"""
seen: set[str] = set()
for column in columns:
problem = _index_column_problem(column, seen)
if problem is not None:
raise ValueError(
f"cannot index resource <{resource_id}>: its search index "
f"field <{column}> {problem}; a field name -- and so every "
f"'meta.labels' key -- must match {INDEX_COLUMN_PATTERN}, "
f"must be neither an SQL keyword nor a name FTS5 reserves "
f"({', '.join(sorted(_INDEX_RESERVED_COLUMNS))}), and must "
f"not repeat another field of the index",
)
seen.add(column.lower())
# One column of the index, as claimed by the first resource that named it:
# its spelling, and the id of that resource.
IndexColumn = tuple[str, str]
# The most columns the repository's index table can have. SQLite's
# SQLITE_MAX_COLUMN is 2000 by default, and FTS5 spends six of those on the
# shadow table behind the virtual one; 1994 is what SQLite 3.53 accepts for
# a CREATE VIRTUAL TABLE ... USING fts5 plus an INSERT naming every column,
# which is what the index build does. The union of a whole repository's
# label keys is what has to fit, so this is checked over the union, not per
# resource (gain#464).
MAX_INDEX_COLUMNS = 1994
[docs]
def merge_index_columns(
resource_id: str,
columns: Sequence[str],
claimed: Mapping[str, IndexColumn],
) -> dict[str, IndexColumn]:
"""Return ``claimed`` extended with ``columns``, keyed case-insensitively.
The index table has one set of columns for the whole repository -- the
union of every resource's fields -- so a field name that is fine within
one resource can still be unusable next to another resource's. SQLite
compares column names case-insensitively, so ``assay`` in one resource
and ``Assay`` in another are one column asked for twice under two
spellings, and a ``CREATE VIRTUAL TABLE`` naming both fails -- taking
the whole repository's index with it (gain#464). Two resources
spelling a field the same way share the column, which is the point of
the index.
Raises ``ValueError`` if ``columns`` cannot join ``claimed`` -- naming
the resource that already holds a spelling, or, past
``MAX_INDEX_COLUMNS``, saying that the repository's labels no longer
fit an FTS5 table. The caller skips that one resource and keeps the
rest. ``claimed`` is never modified -- a rejected resource claims
nothing.
"""
additions: dict[str, IndexColumn] = {}
for column in columns:
key = column.lower()
held = claimed.get(key) or additions.get(key)
if held is None:
additions[key] = (column, resource_id)
continue
spelling, holder = held
if spelling == column:
continue
raise ValueError(
f"cannot index resource <{resource_id}>: its search index "
f"field <{column}> differs only in case from field "
f"<{spelling}> of resource <{holder}>, which the index already "
f"has; SQLite compares column names case-insensitively, so the "
f"two cannot both be fields of the repository's index -- spell "
f"them the same way. Resources join the index in resource id "
f"order, so the spelling of the first resource by id is the "
f"one kept",
)
merged = {**claimed, **additions}
if len(merged) > MAX_INDEX_COLUMNS:
raise ValueError(
f"cannot index resource <{resource_id}>: its search index "
f"fields would take the repository's index past the "
f"{MAX_INDEX_COLUMNS} columns an FTS5 table can have -- the "
f"repository has too many distinct 'meta.labels' keys between "
f"all of its resources. Resources join the index in resource "
f"id order, so the resources dropped are the ones that reach "
f"it last by id",
)
return merged
[docs]
class DerivedFilesState(enum.Enum):
"""Whether a kind's derived files describe the resource as it is now.
A derived file is one a kind computes at repair from inputs the
statistics hash must not learn about, under a freshness gate of its
own; a genomic score keeps its chromosome lengths this way. The gate
answers one of three states and never derives anything.
"""
CURRENT = enum.auto()
"""Present and derived from the resource as it is now."""
STALE = enum.auto()
"""Absent or out of date, and every input to rebuild it is here."""
PAYLOAD_ABSENT = enum.auto()
"""Absent or out of date, and an input is a ``.dvc`` sidecar whose
payload has not been pulled: nothing can be rebuilt here, and there
is nothing wrong with the resource."""
[docs]
class ResourceStatistics:
"""
Base class for statistics.
Subclasses should be created using mixins defined for each statistic type
that the resource contains.
"""
def __init__(self, resource_id: str):
self.resource_id = resource_id
[docs]
@staticmethod
def get_statistics_folder() -> str:
return "statistics"
[docs]
class GenomicResourceImplementation(ABC):
"""Base class of the implementation of a resource type.
A resource is a folder in a repository. An implementation knows how
to describe the resource, how to compute its statistics and how to
render its HTML pages. A subclass overrides the five abstract
methods and, when it reads files, the ``files`` property.
Parameters
----------
genomic_resource : GenomicResource
The resource that the implementation wraps.
Attributes
----------
resource : GenomicResource
The wrapped resource.
config : dict
The configuration read from the resource when the implementation
is constructed. A subclass that validates its configuration may
replace it with the validated form.
"""
def __init__(self, genomic_resource: GenomicResource):
self.resource = genomic_resource
self.config: dict = self.resource.get_config()
self._statistics: ResourceStatistics | None = None
@property
def resource_id(self) -> str:
"""The id of the resource that the implementation wraps."""
return self.resource.resource_id
[docs]
def get_config(self) -> dict:
"""Return the configuration of the resource.
Returns
-------
dict
The configuration read when the implementation was
constructed. A subclass that validates its configuration
returns the validated form.
"""
return self.config
@property
def files(self) -> set[str]:
"""The names of the resource files that the implementation reads."""
return set()
[docs]
@abstractmethod
def calc_statistics_hash(self) -> bytes:
"""Compute the digest of the inputs of the statistics build.
``grr_manage`` compares the digest with the stored one to decide
whether it must compute the statistics again.
Returns
-------
bytes
The digest of everything that the statistics tasks read.
"""
raise NotImplementedError
def stored_statistics(self) -> list[StoredStatistic]:
"""List the versioned statistics files that a build writes.
:meta private:
"""
# The repair flow compares the stored files of a resource whose
# statistics hash is current against this list. It reports a
# missing or older file and never rebuilds it. The list is not
# part of the statistics hash. A kind that declares nothing is
# never reported.
return []
[docs]
@abstractmethod
def create_statistics_build_tasks(
self, *,
region_size: int = DEFAULT_STATISTICS_REGION_SIZE,
grr: GenomicResourceRepo | None = None,
) -> list[TaskDesc]:
"""Create the task graph tasks that compute the statistics.
``grr_manage`` calls every kind with the same two keyword-only
arguments. An implementation declares both, even when it ignores
them, and declares no other argument.
Parameters
----------
region_size : int
The longest region that one task covers. A kind that splits
its build by genomic region reads it. ``0`` means do not
split: a genomic score then scans the resource in one pass
and a reference genome uses one region for each chromosome.
A kind that does not split ignores it.
grr : GenomicResourceRepo | None
The repository that holds the resource. ``None`` means that
the caller names no repository.
Returns
-------
list[TaskDesc]
The tasks of the build. An empty list means that the kind
has no statistics to compute.
Raises
------
ValueError
If ``region_size`` is negative.
"""
raise NotImplementedError
[docs]
@abstractmethod
def calc_info_hash(self) -> bytes:
"""Compute the digest of the inputs of the info page.
The page is rendered again only when the digest changes.
Returns
-------
bytes
The digest of everything that ``get_info`` depends on.
"""
raise NotImplementedError
def derived_files_state(
self, grr: GenomicResourceRepo | None, # ruff: ignore[unused-method-argument]
) -> DerivedFilesState:
"""Report whether the derived files describe the resource now.
:meta private:
"""
# The files are the ones that a kind derives at repair under a
# freshness gate of its own, not under the statistics hash. The
# check is cheap by contract. It compares what a file was derived
# from with what is there now, derives nothing and does not open
# the payload of the resource, so a pointer-only DVC checkout can
# be asked. Most kinds have no such file and answer CURRENT. A
# genomic score keeps its chromosome lengths this way (gain#1576).
return DerivedFilesState.CURRENT
def rebuild_derived_files(
self, grr: GenomicResourceRepo | None, # ruff: ignore[unused-method-argument]
) -> None:
"""Rewrite the files that ``derived_files_state`` gates.
:meta private:
"""
# A repair calls it when the files are STALE while the statistics
# are current. A full statistics build writes them on its own.
# It does nothing for a kind that derives no file.
return
[docs]
@abstractmethod
def get_info(self, **kwargs: Any) -> str:
"""Render the HTML of the info page of the resource.
Parameters
----------
**kwargs : Any
Options of the renderer. The caller defines them.
Returns
-------
str
The HTML of the page.
"""
raise NotImplementedError
[docs]
@abstractmethod
def get_statistics_info(self, **kwargs: Any) -> str:
"""Render the HTML of the statistics page of the resource.
Parameters
----------
**kwargs : Any
Options of the renderer. The caller defines them.
Returns
-------
str
The HTML of the page.
"""
raise NotImplementedError
def collect_index_info(
self,
) -> tuple[tuple[str, ...], tuple[str, ...]]:
"""Collect the resource fields for the full text search index.
:meta private:
"""
# An override that adds fields calls super() and appends to the
# result. This is the only place that checks a label key against
# the names that the index reserves. The re-check of the index
# build sees the finished header, where the fields of an
# implementation legitimately appear, so it cannot tell a field
# from a label (gain#542). A field added by an override belongs in
# GR_INDEX_NON_LABEL_COLUMNS.
#
# Every implementation reaches the index through here, and the
# index build reports a raise from here against this one resource
# (gain#464). A label key that cannot name an index field raises
# ValueError.
res = self.resource
# Through the resource's own narrowing accessor, not the raw
# config: `or {}` guards a None and an empty mapping but not a
# scalar, and a `meta: |` block of prose then raised from the
# `.get` below (gain#1004).
meta = res.get_meta()
labels: dict = res.get_labels() or {}
header: tuple[str, ...] = (
*GR_INDEX_RESOURCE_FIELDS,
*labels.keys(),
)
# Vetted against every non-label column, not just the fields this
# resource's own header carries. A non-score resource carrying a
# label named `score_ids` is the case that reaches this: its own
# header has no score fields, so vetting the header alone lets it
# through (gain#542).
#
# Unconditional on purpose -- do NOT narrow this to "only when the
# repository also holds a score resource". The column such a label
# would need is already the field's, so one row cannot record both
# whatever else the repository holds; and the index's columns are
# the union across the repository, so whether the collision exists
# would otherwise depend on which other resources happen to be
# present. Note this is about what can be *indexed*, not about
# what a query can ask: since gain#646 no label key reaches the
# search statement at all, so a label spelled like a field is
# answered from the resource's own `meta.labels` on both routes.
validate_index_columns(
res.resource_id,
(*sorted(GR_INDEX_NON_LABEL_COLUMNS), *labels.keys()),
)
# Derived from the block narrowed above, through the same helpers
# the resource's own accessors use: the row must carry what
# `get_description`/`get_summary` report, or a resource with a
# description and no summary indexes an empty `summary` column
# while displaying that description as its summary -- invisible to
# a `summary : <term>` search (gain#1008). Off the block rather
# than through the accessors, which would re-narrow it and report a
# malformed one once more per read.
#
# A list value's elements are space-joined, so FTS5 sees one term
# apiece (gain#1225).
row: tuple[str, ...] = (
res.get_full_id(),
res.resource_id,
res.get_type(),
_description_in(meta),
_summary_in(meta),
*[" ".join(label_alternatives(v)) for v in labels.values()],
)
return header, row
[docs]
def get_statistics(self) -> ResourceStatistics | None:
"""Load the statistics of the resource.
Returns
-------
ResourceStatistics | None
The statistics. ``None`` means that the implementation has
none: this base class always returns ``None``.
"""
return None
[docs]
def reload_statistics(self) -> ResourceStatistics | None:
"""Drop the cached statistics and load them again.
Call it after the statistics were rebuilt on disk.
Returns
-------
ResourceStatistics | None
What :meth:`get_statistics` returns. ``None`` means that
the implementation has no statistics.
"""
self._statistics = None
return self.get_statistics()
[docs]
class InfoImplementationMixin:
"""Mixin that provides generic template info page generation interface."""
[docs]
@dataclass
class FileEntry:
"""Provides an entry into manifest object."""
name: str
size: str
md5: str | None
resource: GenomicResource
template_name: ClassVar[str] = "base_implementation.jinja"
styles_template_name: ClassVar[str] = "base_implementation_styles.jinja"
def _get_template_data(self) -> dict:
return {}
[docs]
def get_template_data(self) -> dict:
"""
Return a data dictionary to be used by the template.
Will transform the description in the meta section using markdown.
"""
template_data = self._get_template_data()
template_data["resource_files"] = [
self.FileEntry(entry.name, convert_size(entry.size), entry.md5)
for entry in self.resource.get_manifest().entries.values()
if not entry.name.startswith("statistics")
and entry.name != "index.html"
and not is_dvc_sidecar(entry.name)]
template_data["resource_files"].append(
self.FileEntry("statistics/", "", ""))
# Each label as the strings its value stands for; the template
# joins them for display (gain#1225). Listed by key, not in
# yaml authoring order, so a long row scans by name (gain#1482);
# the alternatives inside a value keep the curator's order.
template_data["labels"] = [
(key, label_alternatives(value))
for key, value in sorted(self.resource.get_labels().items())
]
return template_data
[docs]
def get_statistics_template_data(self) -> dict:
"""
Return a data dictionary to be used by the statistics template.
Will transform the description in the meta section using markdown.
"""
template_data = self._get_template_data()
template_data["statistic_files"] = [
self.FileEntry(
entry.name.removeprefix("statistics/"),
convert_size(entry.size),
entry.md5,
)
for entry in self.resource.get_manifest().entries.values()
if entry.name.startswith("statistics")
and not is_dvc_sidecar(entry.name)]
return template_data
[docs]
def get_info(self, **kwargs: Any) -> str: # ruff: ignore[unused-method-argument]
"""Construct the contents of the implementation's HTML info page."""
template_data = self.get_template_data()
return get_template(self.template_name).render(
resource=self.resource,
data=template_data,
base="resource_template.jinja",
styles_template=self.styles_template_name,
static_root=self._climb_to_root(GR_INDEX_FILE_NAME),
breadcrumb=self._breadcrumb(GR_INDEX_FILE_NAME),
)
[docs]
def get_statistics_info(self, **kwargs: Any) -> str: # ruff: ignore[unused-method-argument]
"""Construct the contents of the implementation's HTML info page."""
template_data = self.get_statistics_template_data()
return get_template(self.template_name).render(
resource=self.resource,
data=template_data,
base="statistics_template.jinja",
styles_template=self.styles_template_name,
static_root=self._climb_to_root(GR_STATISTICS_INDEX_FILE_NAME),
breadcrumb=self._breadcrumb(GR_STATISTICS_INDEX_FILE_NAME),
)
def _climb_to_root(self, page: str) -> str:
"""The ``../`` prefix that takes one of this resource's pages
back to the repository root.
``page`` is the page's path inside the resource directory, as
the publisher names it; the resource directory is the id, one
directory per segment. The templates prefix their urls into
``.static/`` with it (gain#1400).
"""
return climb_to_root(f"{self.resource.resource_id}/{page}")
def _breadcrumb(self, page: str) -> list[Crumb]:
"""The trail from the repository root to one of this resource's
pages, for its header; ``page`` as in :meth:`_climb_to_root`.
"""
return page_breadcrumb(self.resource.resource_id, page)
class _ThreadValidators(threading.local):
"""One cerberus ``Validator`` per implementation type, per thread.
Subclassing ``threading.local`` rather than using one directly is what
makes ``__init__`` run once per thread, so the payload is simply there
on first access instead of being lazily created at every use.
"""
# pylint: disable=too-few-public-methods
def __init__(self) -> None:
self.by_type: weakref.WeakKeyDictionary[type, Validator] = \
weakref.WeakKeyDictionary()
class _Memo(NamedTuple):
"""One resource's normalized config, and the config it came from.
``config`` is held to be compared by *identity* on the next lookup, not
read: it is the object ``document`` was normalized from, so a caller
handing over the same object again is asking the same question. Held
strongly, which normally costs nothing -- it is the resource's own
config, and one dict is retained per (resource, type) regardless.
"""
config: dict
document: dict
class _ConfigValidatorCache:
"""The resource-config schemas, validators, and normalized documents.
A resource's schema and the cerberus ``Validator`` compiled from it
depend on the implementation TYPE, never on the config being validated,
yet both were rebuilt on every call -- and constructing a ``Validator``
makes cerberus validate the schema definition itself. Together that was
about a third of the cost of validating one resource, paid once per
resource, and a wildcard pipeline build validates hundreds (gain#905).
**The normalized document is kept too**, which is what remained after
gain#905 and turned out to be the larger half (gain#1059). It depends
on the config and not only on the type, so it is keyed per (resource,
type) -- and only a run whose tasks share a process collects on that: a
resource that crosses a process boundary is unpickled into a new object,
which is a new key and so a fresh normalization.
That makes this the first store here whose size follows the *repository*
rather than the handful of implementation types: one document per live
resource something has built an implementation for, each weighing about
what that resource's config already weighs (2.06 MiB for the 321
resources of grr plus grr_sfari). The keys are weak, so a resource
nothing holds takes its entry with it -- but a protocol holds every
resource it has enumerated, so a repo-wide walk retains one document per
resource for as long as it runs.
This is not the convention the four ``get_memo_key()`` memos use
(``gene_scores``, ``gene_set``, ``gene_models_factory``,
``liftover_chain``), and deliberately: those memoize an object built
*from* a resource, under a strong string key that never evicts, whereas
what is memoized here is the normalization of a config passed as an
argument -- which callers do supply independently of the resource.
**The validator is deliberately not shared between threads.** Cerberus
keeps per-call state on the instance -- ``validate()`` assigns
``self.document`` and ``self._errors``, and the caller reads the
normalized document back off the instance on the next line -- so one
instance shared across threads can hand a caller another caller's
normalized config. That is not theoretical: pipeline loads run on a
thread pool and loading a pipeline constructs resource implementations,
and a direct reproduction bled one read in 600. Hence per thread; a
thread's validators die with it.
**The schema is shared between threads, for a narrower reason than
"cerberus treats it as read-only" -- it does not.** ``DefinitionSchema``
*expands* the schema in place, rebinding nested rule values, both when a
validator is constructed and again on every ``validate()``. What holds
is that the expansion is value-preserving for the schemas gain actually
has, none of which uses a rule cerberus rewrites non-trivially. The
exposure is not new either: ``get_schema()`` splices in the *same*
module-level ``AGGREGATOR_SCHEMA`` object rather than a copy, so that
fragment was already being re-expanded concurrently on every call before
anything was cached. Being by value is also why the guarding test
asserts ``==`` and not ``is``.
Nothing guards against re-entrancy: a nested validation of the same type
on the same thread would overwrite the outer call's document before it is
read. No such path exists -- no gain schema carries a callable rule
(``check_with``/``coerce``/``default_setter``), and the only code between
``validate()`` and the read is the failure branch's eagerly-evaluated log
call. Adding a callable rule to a resource schema would change that.
Keys are held weakly so that an implementation type defined inside a test
does not outlive it; production types are module-level and immortal
anyway.
"""
def __init__(self) -> None:
self._schemas: weakref.WeakKeyDictionary[type, dict] = \
weakref.WeakKeyDictionary()
self._validators = _ThreadValidators()
# Weak outer, plain inner, as `_REF_GENOME_CACHE` is: an entry's
# implementation type is held only for as long as its resource, so
# a type defined inside a test still dies with it.
self._documents: weakref.WeakKeyDictionary[
GenomicResource, dict[type, _Memo],
] = weakref.WeakKeyDictionary()
def schema_for(self, implementation: type) -> dict:
"""Return ``implementation``'s schema, building it at most once.
Two threads racing to fill an entry both call ``get_schema()`` and
one overwrites the other, which is harmless -- the schemas are equal.
"""
schema = self._schemas.get(implementation)
if schema is None:
schema = implementation.get_schema() # type: ignore[attr-defined]
self._schemas[implementation] = schema
return schema
def validator_for(self, implementation: type) -> Validator:
"""Return this thread's validator for ``implementation``."""
validator = self._validators.by_type.get(implementation)
if validator is None:
# pylint: disable=not-callable
validator = Validator(self.schema_for(implementation))
self._validators.by_type[implementation] = validator
return validator
def document_for(
self, implementation: type, resource: GenomicResource, config: dict,
) -> dict | None:
"""Return ``config`` normalized before, or ``None`` if it was not.
A hit requires the *same config object*, not an equal one -- the
precondition that goes with which is on
``validate_and_normalize_schema``, where a caller will read it.
Keying on the resource alone would be wrong: the same resource is
routinely normalized against more than one config, and every caller
is entitled to the document for the config it actually passed.
Switching the guard to ``==`` would not make it notice an edited
config, which is the tempting thing to think: an entry holds the
caller's config object itself, so equality compares that object with
itself and says yes however it has been edited since.
Keys are compared by value, ``GenomicResource`` having its own
``__eq__``, so a ``CacheResource`` and the resource it mirrors are
one key -- and they also share the one config object, so the hit is
correct rather than merely harmless. A hit pays exactly one such
``__eq__``, comparing a config with itself, which short-circuits per
value on identity: 0.13 us for the largest config in the GRR.
"""
by_type = self._documents.get(resource)
if by_type is None:
return None
memo = by_type.get(implementation)
if memo is None or memo.config is not config:
return None
return copy.deepcopy(memo.document)
def remember_document(
self, implementation: type, resource: GenomicResource,
config: dict, document: dict,
) -> None:
"""Memoize ``document`` as the normalization of ``config``.
Stored as a copy, so that an entry is a value rather than a view:
cerberus rebuilds only the mappings its schema describes, so
``document`` still holds the resource's own objects for fields like
``meta.labels``. Callers are handed copies too, so nothing can
observe the difference -- the copy buys the invariant, at 1.8% of a
cold pass over a whole GRR, paid once per entry.
As with schemas, two threads may both fill an entry and one overwrite
the other; the documents are equal. No lock, on either path. A
lost or duplicated entry is a miss and never a wrong answer, since a
hit is rechecked against the config it was filled from -- which is
also what makes the unguarded weakref removals this store provokes
(its keys die, unlike the immortal types beside it) harmless on the
3.14t lane ``Jenkinsfile.python-matrix`` runs.
"""
by_type = self._documents.setdefault(resource, {})
by_type[implementation] = _Memo(config, copy.deepcopy(document))
def clear(self) -> None:
"""Forget every schema and document, and this thread's validators.
Exists for tests. The cache is process-wide and never invalidated,
which is right for a running gain -- the schema of a type does not
change -- and wrong across tests, which share one process and one
main thread: a validator built while ``Validator`` was monkeypatched
outlives the patch and would answer for a later test.
Only the calling thread's validators are forgotten. Another live
thread's are not reachable from here, and a registry that made them
so would put a lock on the validation path to serve a test.
"""
self._schemas.clear()
self._validators.by_type.clear()
self._documents.clear()
#: The process-wide resource-config validator cache. See the class.
CONFIG_VALIDATOR_CACHE = _ConfigValidatorCache()
[docs]
class ResourceConfigValidationMixin:
"""Mixin that provides validation of resource configuration."""
[docs]
@staticmethod
@abstractmethod
def get_schema() -> dict:
"""Return schema to be used for config validation."""
raise NotImplementedError
[docs]
@classmethod
def validate_and_normalize_schema(
cls, config: dict, resource: GenomicResource) -> dict:
"""Validate the resource schema and return the normalized version.
What comes back is the caller's own document all the way down, memo
hit or not, and detached from the config it was normalized from -- so
an implementation may keep it and write into it, as
``GenomicScore.__init__`` does.
**Do not edit a resource's config in place.** Offering the same
config object twice is answered from a memo rather than re-normalized
(gain#1059), and an edit made between the two calls is not seen: the
second caller gets the document as the config was the first time.
Nothing in gain or gpf does this -- the implementations that validate
all keep the normalized copy rather than the config -- and code that
wants a config re-read should hand over a new dict, which is always
normalized afresh.
"""
memoized = CONFIG_VALIDATOR_CACHE.document_for(cls, resource, config)
if memoized is not None:
return memoized
validator = CONFIG_VALIDATOR_CACHE.validator_for(cls)
if not validator.validate(config):
logger.error(
"Resource %s of type %s has an invalid configuration. %s",
resource.resource_id,
resource.get_type(),
validator.errors)
# A config refusal is the resource's own fault (ADR 0008,
# decision 6): the type puts it in `RESOURCE_ERRORS`, and the
# prefix is the one `resource_errors` builds for a score's
# definition, so a reader sees one wording for both.
raise MalformedResourceError(
f"Invalid configuration: {resource.resource_id}")
document = cast("dict", validator.document)
CONFIG_VALIDATOR_CACHE.remember_document(
cls, resource, config, document)
# Copied even here, where the document is cerberus's own and freshly
# made. Cerberus copies the config shallowly and rebuilds only the
# mappings its schema describes, so a field with no sub-schema --
# `meta.labels`, `default_annotation` -- comes back as the very
# object in the resource's config. Handing that to the caller who
# missed, and a detached copy to everyone who hits, would make
# "whose dict is this" depend on cache state.
return copy.deepcopy(document)