policy_specs#

class ContainerRole(value)[source]#

Bases: StrEnum

The role of the instance container in the overall config structure. 1) SOURCE_OF_TRUTH: authoritative source of truth for the given keys. (typically for sample.channels) 2) INSTANCE_RUNTIME: runtime instance container, derived from source of truth.

INSTANCE_RUNTIME = 'instance_runtime'#
SOURCE_OF_TRUTH = 'source_of_truth'#
class ContainerSpecRegistry(specs: Iterable[InstanceContainerSpec])[source]#

Bases: object

all_instance_container_abs_paths() tuple[Tuple[str, ...], ...][source]#
by_template_kind() dict[tuple[str, TemplateKind], InstanceContainerSpec][source]#

Map (section, template_kind) -> spec.

excluded_roots(section: str) set[Tuple[str, ...]][source]#

Excluded subtrees for STATIC fillers within a section (section-relative paths): - instance containers (runtime) - templates reservoirs

for_section(section: str) tuple[InstanceContainerSpec, ...][source]#
generic_instance_specs() tuple[InstanceContainerSpec, ...][source]#
instance_roots(section: str) set[Tuple[str, ...]][source]#
ordered_for_reconcile(*, kind: ReconcileKind = ReconcileKind.GENERIC) tuple[InstanceContainerSpec, ...][source]#
Ordering for reconcile execution:
  1. role: SOURCE_OF_TRUTH before INSTANCE_RUNTIME

  2. mode: FILL_MISSING_ONLY before REPLACE_CONTAINER

  3. stable tie-breakers: section, container_path

skip_paths_for_missing_only_merge(section: str) set[Tuple[str, ...]][source]#
special_instance_specs() tuple[InstanceContainerSpec, ...][source]#
template_roots(section: str) set[Tuple[str, ...]][source]#
property specs: tuple[InstanceContainerSpec, ...]#
class InstanceContainerSpec(section: 'str', container_path: 'str', defaults_templates_path: 'str', template_kind: 'TemplateKind', membership: 'Membership', role: 'ContainerRole' = <ContainerRole.INSTANCE_RUNTIME: 'instance_runtime'>, rename: 'RenameStepSpec' = RenameStepSpec(applies=True, migrate_payload=True), reconcile: 'ReconcileStepSpec' = ReconcileStepSpec(applies=True, mode=<ReconcileMode.REPLACE_CONTAINER: 'replace_container'>, kind=<ReconcileKind.GENERIC: 'generic'>, policy=InstancePolicy(preserve_existing=True, prune_removed=True), preserve_existing_order=True, restrict_to_template_keys=False), compound: 'bool' = False, compound_oriented: 'bool' = False, compound_prune_invalid_atoms: 'bool' = True, ctx_builder: "Optional[Callable[[str, dict, 'SampleManagerProtocol'], Dict[str, object]]]" = None)[source]#

Bases: object

abs_container_path() Tuple[str, ...][source]#
abs_templates_path() Tuple[str, ...] | None[source]#
container_map(view: Mapping[str, Any]) dict[str, Any][source]#
owns_prefix() Tuple[str, ...][source]#
rel_container_path() Tuple[str, ...][source]#
rel_templates_path() Tuple[str, ...] | None[source]#
resolve_keys(*, sm, resolver) list[str][source]#
compound: bool = False#
compound_oriented: bool = False#
compound_prune_invalid_atoms: bool = True#
container_path: str#
ctx_builder: Callable[[str, dict, SampleManagerProtocol], Dict[str, object]] | None = None#
defaults_templates_path: str#
membership: Membership#
reconcile: ReconcileStepSpec = ReconcileStepSpec(applies=True, mode=<ReconcileMode.REPLACE_CONTAINER: 'replace_container'>, kind=<ReconcileKind.GENERIC: 'generic'>, policy=InstancePolicy(preserve_existing=True, prune_removed=True), preserve_existing_order=True, restrict_to_template_keys=False)#
rename: RenameStepSpec = RenameStepSpec(applies=True, migrate_payload=True)#
role: ContainerRole = 'instance_runtime'#
section: str#
template_kind: TemplateKind#
class InstancePolicy(preserve_existing: bool = True, prune_removed: bool = True)[source]#

Bases: object

Policy for handling existing keys in instance containers.

Parameters:
  • preserve_existing (bool) – If True, existing keys in the instance container are preserved (not overwritten). If False, existing keys are overwritten with new values from the template.

  • prune_removed (bool) – If True, keys that are no longer present in the template are removed from the instance container

preserve_existing: bool = True#
prune_removed: bool = True#
class Membership(source: Literal['sample', 'pipeline'], pipeline: str | None = None, instance_kind: str | None = None, oriented: bool = False)[source]#

Bases: object

Defines the target keyspace for an instance container. The spec is declarative; keys() is the single execution point.

keys(sm, resolver) list[str][source]#
instance_kind: str | None = None#
oriented: bool = False#
pipeline: str | None = None#
source: Literal['sample', 'pipeline']#
class ReconcileKind(value)[source]#

Bases: StrEnum

GENERIC = 'generic'#
SPECIAL = 'special'#
class ReconcileMode(value)[source]#

Bases: StrEnum

How to apply changes to existing instance containers. 1) REPLACE_CONTAINER: fully replace the container with the new materialized one (including removing keys no longer present). 2) FILL_MISSING_ONLY: only fill in missing keys, do not remove existing keys or overwrite present ones.

FILL_MISSING_ONLY = 'fill_missing_only'#
REPLACE_CONTAINER = 'replace_container'#
class ReconcileStepSpec(applies: 'bool' = True, mode: 'ReconcileMode' = <ReconcileMode.REPLACE_CONTAINER: 'replace_container'>, kind: 'ReconcileKind' = <ReconcileKind.GENERIC: 'generic'>, policy: 'InstancePolicy' = InstancePolicy(preserve_existing=True, prune_removed=True), preserve_existing_order: 'bool' = True, restrict_to_template_keys: 'bool' = False)[source]#

Bases: StepSpec

kind: ReconcileKind = 'generic'#
mode: ReconcileMode = 'replace_container'#
policy: InstancePolicy = InstancePolicy(preserve_existing=True, prune_removed=True)#
preserve_existing_order: bool = True#
restrict_to_template_keys: bool = False#
class RenameStepSpec(applies: 'bool' = True, migrate_payload: 'bool' = True)[source]#

Bases: StepSpec

migrate_payload: bool = True#
class SmellScanPolicy(strong_template_smells: 'bool' = True, light_channelish_smells: 'bool' = True, reserved_namespace_keys: 'Tuple[str, ...]' = ('performance',), templateish_exact: 'Tuple[str, ...]' = ('templates', 'channel', 'layout_template', 'derived_template'), templateish_suffix: 'str' = '_template')[source]#

Bases: object

light_channelish_smells: bool = True#
reserved_namespace_keys: Tuple[str, ...] = ('performance',)#
strong_template_smells: bool = True#
templateish_exact: Tuple[str, ...] = ('templates', 'channel', 'layout_template', 'derived_template')#
templateish_suffix: str = '_template'#
class StepSpec(applies: bool = True)[source]#

Bases: object

Base class for step specifications.

applies: bool = True#
class TemplateKind(value)[source]#

Bases: StrEnum

The semantic kind of template to use for instance container materialization. 1) CHANNELS: per-channel templates (e.g. registration.channels) 2) PAIRS: per-pair templates (e.g. colocalization.pairs) 4) PERF: per-channel performance templates (e.g. performance.binarization.channels)

CHANNELS = 'channels'#
PAIRS = 'pairs'#
PERF = 'perf'#