"""
Providers and the chain
=======================
A provider answers part queries. A :py:class:`PartSource` is an ordered chain of
them, asked in order, first hit wins.
To write one, subclass :py:class:`Provider`, or :py:class:`CategoryProvider` to
answer one query at a time rather than a batch, and implement three methods:
`config()`, which identifies the provider so a chain has a stable digest;
`resolve()`, which answers a batch; and `rehydrate()`, which rebuilds a part from
a lockfile entry this provider wrote. `distinct()` is optional and serves
exploration only. Everything else has a default.
An answer is per query, not per provider. :py:class:`Resolved` retires the query,
:py:class:`NoMatch` and :py:class:`Unsupported` let the chain continue, and
:py:class:`Deny` stops it. A provider is asked every query in its chain and
declines what it cannot serve, per query, by returning `Unsupported` with the keys
it could not process -- there is no capability manifest, so nothing can skip a
provider on a declaration, and the aggregated diagnostic can say that nothing in
the chain serves `tcr` as opposed to nothing matching it.
`resolve` is handed a batch rather than a query so a provider can answer many at
once, but **a batch is not the design**: only queries the lock could not answer
reach the chain, so it is the whole set only on a first run, under `off`, or in
`check`'s second call -- and duplicates are collapsed onto
:py:attr:`CanonicalQuery.count`. Anything order-wide -- order quantities, rebates,
reel consolidation -- has to be written knowing that. See
:py:meth:`Provider.resolve`.
A `PartSource` is a :py:class:`jitx.context.Context`, so `with PartSource([...]):`
scopes it and a `Design` can carry one as an attribute. Its members may be
provider instances or registry strings; see :py:mod:`jitxlib.parts.policy` for
where a chain comes from when the design supplies none.
This module also re-exports the query and record names a provider needs, so a
provider has one import site rather than four.
"""
from __future__ import annotations
import json
from abc import ABC, abstractmethod
from collections.abc import Iterator, Mapping, Sequence
from dataclasses import dataclass, replace
from typing import Any, TypeAlias
import jitx.context
from .error import (
PartDenied,
PartNotFound,
ProviderContractError,
)
# Re-exported for provider authors, so a provider has one import site rather than
# four. The `X as X` form is what marks a re-export as intentional; a bare import
# would need a lint suppression to say the same thing less clearly.
from .query import (
AtLeast as AtLeast,
AtMost as AtMost,
CanonicalQuery as CanonicalQuery,
Exactly as Exactly,
Exists as Exists,
OneOf as OneOf,
Predicate as Predicate,
Range as Range,
Rank as Rank,
Scalar as Scalar,
Tolerance as Tolerance,
is_quantity as is_quantity,
match as match,
rank_by as rank_by,
require_dimension as require_dimension,
satisfies as satisfies,
satisfies_or_true as satisfies_or_true,
)
from .lock import (
decode_payload as decode_payload,
encode_payload as encode_payload,
)
from .record import (
JSON as JSON,
RESERVED_FIELDS as RESERVED_FIELDS,
PartIdentity as PartIdentity,
ResolvedPart as ResolvedPart,
apply_fields as apply_fields,
)
# --------------------------------------------------------------------------
# Per-query outcomes
# --------------------------------------------------------------------------
[docs]
@dataclass(frozen=True)
class Resolved:
"""Parts this provider offers, ranked best-first.
Never empty: a provider with nothing to offer returns `NoMatch`. Selection is
`parts[0]`, so an empty tuple would retire a query with nothing to bind.
"""
parts: tuple[ResolvedPart, ...]
def __post_init__(self):
if not self.parts:
raise ProviderContractError(
"Resolved.parts is empty; a provider with nothing to offer "
"returns NoMatch(), not Resolved(())"
)
[docs]
@dataclass(frozen=True)
class NoMatch:
"""Understood the query, has nothing. The chain continues."""
reason: str = ""
[docs]
@dataclass(frozen=True)
class Unsupported:
"""Cannot honor part of the query. The chain continues.
`keys` names the query keys this provider cannot process, which is what lets
the aggregated diagnostic tell a design author that nothing in the chain
serves `tcr`, as opposed to nothing matching.
"""
keys: tuple[str, ...] = ()
reason: str = ""
[docs]
@dataclass(frozen=True)
class Deny:
"""Policy veto. The chain stops and resolution fails with this reason."""
reason: str
Outcome: TypeAlias = Resolved | NoMatch | Unsupported | Deny
[docs]
@dataclass(frozen=True)
class ChainOutcome:
"""The chain's answer for one query, plus what each member said on the way.
One `Outcome` cannot carry per-provider accumulation, and the diagnostics need
all of it.
"""
outcome: Outcome
trail: tuple[tuple[str, Outcome], ...] = ()
[docs]
def describe(outcome: Outcome) -> str:
"""One-line rendering of an outcome, for diagnostics."""
if isinstance(outcome, Resolved):
return f"{len(outcome.parts)} candidate(s)"
if isinstance(outcome, NoMatch):
return outcome.reason or "no match"
if isinstance(outcome, Unsupported):
if outcome.reason:
return f"unsupported: {outcome.reason}"
return f"unsupported: {', '.join(outcome.keys) or 'query'}"
if isinstance(outcome, Deny):
return f"denied: {outcome.reason}"
return repr(outcome)
# --------------------------------------------------------------------------
# The provider protocol
# --------------------------------------------------------------------------
def _canonical(value: object) -> str:
"""A canonical rendering of a JSON value, ordered so spelling does not matter.
Raises if the value is not JSON-renderable, which for a `config()` is a
provider contract violation.
"""
# `object` rather than `JSON`, since a list of configs is not a `list[JSON]`:
# lists are invariant. No `default=`, so a bad value raises here instead of
# making the digest depend on a repr.
return json.dumps(value, sort_keys=True, separators=(",", ":"))
[docs]
class Provider(ABC):
"""What a parts provider implements.
Subclass this, or `CategoryProvider` to answer one query at a time. `config`,
`resolve` and `rehydrate` are yours to write; everything else has a default.
"""
name: str
"""How this provider is named in diagnostics and lockfile entries.
Supply it as a dataclass field, a plain class attribute, or an assignment in
`__init__`. A `property` or a `ClassVar` is an incompatible override; for a
derived name, use `field(init=False)` and set it in `__post_init__`.
"""
discovery_priority: int = 0
"""Rank used ONLY when no chain is declared anywhere and one is discovered
from entry points; higher is asked first. An explicit `PartSource` ignores it,
because a written chain is already ordered.
Declare it the same way as `name`.
"""
[docs]
@abstractmethod
def config(self) -> dict[str, JSON]:
"""This provider's configuration as a canonical JSON object.
An object rather than a bare value, so a config is always something with
named keys, which is what makes two of them comparable and what keeps a
chain digest readable in the file it lands in.
Must cover everything that could change which part gets selected (a file
path, a base URL, a series name) and nothing that could not, such as an
open connection or a warmed cache.
"""
[docs]
@abstractmethod
def resolve(self, batch: Sequence[CanonicalQuery]) -> Sequence[Outcome]:
"""Answer a batch. Index-aligned with `batch`: one outcome per query.
**The batch is not the design.** It carries the queries that reached the
chain on this flush, which is all of them only when all of them are new:
a first run against an empty lock, or `off` mode. Under `read` and
`read-write` a query the lock can answer never gets here, so against a
populated lockfile the batch is the *unpinned remainder* -- often empty.
`replay` never calls this at all. Batching is best effort on top of that,
so an early flush splits one elaboration into several batches and no batch
can be assumed to be the whole of even the remainder.
`check` is the exception and always asks twice per flush: once for the
ordinary run, over the unpinned remainder, which is what it binds, and
once over *every* query the design asked, which is the clean run it
compares the board against. The second is the whole order, so a provider
doing order-level work is asked the same question a from-scratch run would
ask -- and may answer it differently from the first without contradicting
itself.
Duplicates are collapsed: one question is one entry however many
components asked it, since a hundred identical capacitors must not become
a hundred catalogue scans. `CanonicalQuery.count` carries how many asked.
A provider doing whole-batch optimization -- minimum order quantities,
bulk rebates, consolidating a reel, allocation across sources -- has to be
written knowing all of that. Those are properties of the whole order, and
the batch is the whole order only on a run where nothing is pinned. The
tension is inherent rather than an implementation gap: if a selection
depends on the order, adding one component can legitimately move every
selection, which is the thing a lock exists to prevent. A provider that
makes that trade owns it.
"""
[docs]
@abstractmethod
def rehydrate(
self, identity: PartIdentity, payload: dict[str, JSON]
) -> ResolvedPart:
"""Reconstruct a part from a lockfile entry this provider wrote.
Required of any provider used with a lockfile: only the provider knows
what `payload` means. May raise `StalePayload` if the payload predates a
change it cannot honor.
"""
def __eq__(self, other: object) -> bool:
"""Same type, same configuration."""
if not isinstance(other, Provider) or type(other) is not type(self):
return False
return other.config() == self.config()
def __hash__(self) -> int:
"""Over the canonical form of `config()`, which a JSON value is not."""
return hash(_canonical(self.config()))
[docs]
def distinct(self, query: CanonicalQuery, key: str) -> Sequence[object] | None:
"""Values of `key` among parts matching `query`. Exploration only.
Optional, and `None` is how a provider says it does not enumerate --
which is distinct from `()`, meaning it enumerated and found nothing. The
default says the former, so a provider with no business answering need not
mention it.
Not batched, unlike `resolve`: it is called from a REPL one question at a
time, so the batch-is-canonical rule does not apply. There is no fallback
and no `Deny`, because nothing is being selected.
"""
return None
[docs]
class CategoryProvider(Provider):
"""Convenience base for a provider that answers per query rather than per batch.
Implements `resolve` by looping `resolve_one`, which is what a provider with
no batch advantage wants. A provider that can exploit the batch, say one round
trip for three hundred queries, should implement `resolve` directly -- and
read what `Provider.resolve` says about what a batch does and does not contain
before relying on it for anything order-wide.
"""
[docs]
@abstractmethod
def resolve_one(self, query: CanonicalQuery) -> Outcome:
"""Answer one query. `config` and `rehydrate` are still yours to write."""
[docs]
def resolve(self, batch: Sequence[CanonicalQuery]) -> Sequence[Outcome]:
return [self.resolve_one(q) for q in batch]
# --------------------------------------------------------------------------
# The chain
# --------------------------------------------------------------------------
[docs]
@dataclass(frozen=True)
class Named:
"""Give a chain member an explicit name.
A member's name is what a lockfile entry records and what replay looks up, so
two members answering to the same name have to be told apart. They are
suffixed positionally by default -- `acme`, `acme#2` -- which works but says
nothing, and moves if the chain is reordered. Name them instead::
PartSource([
Named("approved", AcmeParts(APPROVED_DSN)),
Named("fallback", AcmeParts(FALLBACK_DSN)),
])
Not a provider and not a wrapper around one: it is a declaration `PartSource`
reads and discards, so it adds no layer for a part to be resolved through. A
member's name is part of the chain's identity, so renaming one re-keys the
chain and its entries re-resolve once, deliberately, like any other chain
edit.
"""
name: str
provider: Provider | str
[docs]
class PartSource(jitx.context.Context):
"""An ordered chain of providers.
Providers are checked in the order given, first hit wins. `label` names this
chain in lockfile entries and diagnostics; it is cosmetic and derived if
omitted. Wrap a member in :py:class:`Named` to give it an explicit name,
which is worth doing whenever two members would otherwise collide.
Strings in the chain are resolved through the entry-point registry the first
time the chain is used, not at construction, so naming a provider package you
have not installed fails with a clear error later rather than at import time.
"""
# Not a dataclass, deliberately. It needs a real __init__, to accept any
# Sequence, and to hold the lazily-filled resolution cache, which is not a
# value and must not reach __eq__ or config(), and a frozen dataclass with
# a custom __init__ is just object.__setattr__ with extra steps. Immutable by
# discipline instead: nothing here reassigns after construction, and
# __eq__/__hash__ are written out below.
providers: tuple[Provider | str, ...]
label: str | None
def __init__(
self,
providers: Sequence[Provider | str | Named] | None = None,
label: str | None = None,
):
members = tuple(providers or ())
self.providers = tuple(
m.provider if isinstance(m, Named) else m for m in members
)
self.given_names = tuple(
m.name if isinstance(m, Named) else None for m in members
)
"""The explicit name of each member, or None where it takes its own."""
self.label = label
self._resolved: tuple[Provider, ...] | None = None
def __repr__(self) -> str:
members = [
Named(given, p) if given is not None else p
for p, given in zip(self.providers, self.given_names, strict=True)
]
return f"PartSource({members!r}, label={self.label!r})"
# -- string resolution --------------------------------------------------
def _resolve_members(self) -> tuple[Provider, ...]:
"""Resolve string members to instances, once, and cache the tuple."""
if self._resolved is not None:
return self._resolved
from .policy import resolve_provider_name
self._resolved = tuple(
resolve_provider_name(p) if isinstance(p, str) else p
for p in self.providers
)
return self._resolved
def __iter__(self) -> Iterator[Provider]:
"""Yields resolved providers, never the strings a chain may be written with.
So composition produces instances, and `__eq__`/`__hash__`/`config()` all
see what iteration does. Iterating therefore resolves.
"""
return iter(self._resolve_members())
def __len__(self) -> int:
return len(self.providers)
def __eq__(self, other: object) -> bool:
if not isinstance(other, PartSource):
return NotImplemented
return self._resolve_members() == other._resolve_members()
def __hash__(self) -> int:
return hash(self._resolve_members())
[docs]
def config(self) -> list[dict[str, JSON]]:
"""Ordered configs of the chain; the chain digest is built from this.
The *member* name, not the provider's own: how a member is addressed is
part of the chain's identity, since that is what lock entries reference.
Renaming one is a chain edit and re-keys accordingly.
"""
return [
{"name": name, "config": p.config()}
for p, name in zip(self, self.names(), strict=True)
]
[docs]
def digest(self) -> str:
"""Stable identity of this chain, across runs and machines."""
import hashlib
return hashlib.sha256(_canonical(self.config()).encode("utf-8")).hexdigest()
[docs]
def names(self) -> list[str]:
"""How each member is addressed: what a lock entry records.
An explicit `Named` wins; otherwise the provider's own name, positionally
suffixed where two collide. Duplicate names in a Python-built chain are
legal, so anything keyed by name has to disambiguate rather than silently
merge two providers -- but a positional suffix says nothing about which
member is which and moves when the chain is reordered, so an ambiguous
chain is better off naming its members.
Two explicit names that collide are an error, not a thing to suffix: the
point of giving one is to be unambiguous.
"""
given = [name for name in self.given_names if name is not None]
clash = next((n for n in given if given.count(n) > 1), None)
if clash is not None:
raise ProviderContractError(
f"two chain members are both named {clash!r}; an explicit name "
f"has to be unique within its chain"
)
taken = set(given)
seen: dict[str, int] = {}
out: list[str] = []
for provider, explicit in zip(self, self.given_names, strict=True):
if explicit is not None:
out.append(explicit)
continue
name = provider.name
if name in seen:
seen[name] += 1
name = f"{name}#{seen[name]}"
else:
seen[name] = 1
if name in taken:
raise ProviderContractError(
f"chain member {name!r} collides with a name given "
f"explicitly to another member"
)
out.append(name)
return out
# -- resolution ---------------------------------------------------------
[docs]
def outcomes(self, batch: Sequence[CanonicalQuery]) -> Sequence[ChainOutcome]:
"""Per-query results, index-aligned. Never raises for a query-level result.
This is what a delegating provider calls: `resolve` raises on exhaustion,
which would make its own `NoMatch` and `Deny` branches unreachable.
"""
pending = list(range(len(batch)))
trails: list[list[tuple[str, Outcome]]] = [[] for _ in batch]
finals: list[Outcome | None] = [None for _ in batch]
for provider, name in zip(self, self.names(), strict=True):
if not pending:
break
sub = [batch[i] for i in pending]
answers = list(provider.resolve(sub))
if len(answers) != len(sub):
raise ProviderContractError(
f"{name} returned {len(answers)} outcomes for {len(sub)} "
f"queries; resolve must be index-aligned with its batch"
)
still: list[int] = []
for index, answer in zip(pending, answers, strict=True):
trails[index].append((name, answer))
if isinstance(answer, (Resolved, Deny)):
finals[index] = answer
else:
still.append(index)
pending = still
return [
ChainOutcome(
outcome=final if final is not None else NoMatch("chain exhausted"),
trail=tuple(trail),
)
for final, trail in zip(finals, trails, strict=True)
]
[docs]
def resolve(self, batch: Sequence[CanonicalQuery]) -> Sequence[ResolvedPart]:
"""The resolver's entry point. Wraps `outcomes` and raises.
`PartNotFound` on exhaustion, `PartDenied` on a veto. Note which method
the last step belongs to: `outcomes` reports exhaustion, `resolve` raises
on it. Conflating the two makes delegation impossible.
Use `selections` instead where the answering member matters, which for
anything writing a lockfile it does.
"""
return [part for part, _ in self.selections(batch)]
[docs]
def selections(
self, batch: Sequence[CanonicalQuery]
) -> Sequence[tuple[ResolvedPart, str]]:
"""Each query's part and **which member answered it**, index-aligned.
Raises exactly as `resolve` does; it is the same call with the one fact
`resolve` throws away. The member is its name from `names()`, which is
what a lock entry records and what replay looks up.
That fact cannot be recovered afterwards. `part.identity.provider` is
free text the provider chose: two members may answer to one name, and a
delegating provider names its *inner* source, which is not a member at
all. Only the chain knows, and only here.
"""
names = self.names()
out: list[tuple[ResolvedPart, str]] = []
for query, chain_outcome in zip(batch, self.outcomes(batch), strict=True):
outcome = chain_outcome.outcome
if isinstance(outcome, Resolved):
member = next(
(
name
for name, answer in reversed(chain_outcome.trail)
if isinstance(answer, Resolved)
),
names[0] if names else "",
)
out.append((outcome.parts[0], member))
continue
if isinstance(outcome, Deny):
who = next(
(n for n, o in chain_outcome.trail if isinstance(o, Deny)),
"a provider",
)
raise PartDenied(who, outcome.reason, query.describe())
raise PartNotFound(
query.describe(),
[(name, describe(o)) for name, o in chain_outcome.trail]
or [("(chain)", "the chain is empty")],
)
return out
[docs]
def candidates(
self, query: CanonicalQuery, limit: int
) -> dict[str, list[ResolvedPart]]:
"""Up to `limit` candidates, keyed by the member that offered them.
Browsing, not choosing, and the two want different things from a chain.
`resolve` takes the first member that answers and stops, which is what
precedence means when one part is being *selected*. A `limit` asks for up
to N, so stopping at a member that offered three of them answers with
three while the rest of the chain holds hundreds -- which is not what was
asked.
Keyed by member, like `distinct`, and for its reason: which catalogue has
a part is usually what the caller is trying to find out. The mapping is in
chain order, so precedence is still readable, and nothing is compared
*across* members -- ranking two providers' parts against each other needs
an agreed vocabulary for `price` and the rest that this design
deliberately does not impose.
**Only members actually asked appear.** The walk stops once `limit` is
filled, and a member that was never reached is absent rather than present
with an empty list, which would claim it had been asked and had nothing.
A member that *was* asked and had nothing is present with `[]`.
Each member is asked only for the shortfall, and truncated to it, so a
provider that ignores `limit` cannot push the total over it.
A `Deny` stops the walk and raises, as in `resolve`: a veto says the query
should not be asked at all, and listing a later member's candidates would
defeat it.
"""
out: dict[str, list[ResolvedPart]] = {}
found = 0
for provider, name in zip(self, self.names(), strict=True):
if found >= limit:
break
answers = list(provider.resolve([replace(query, limit=limit - found)]))
if len(answers) != 1:
raise ProviderContractError(
f"{name} returned {len(answers)} outcomes for 1 query; "
f"resolve must be index-aligned with its batch"
)
outcome = answers[0]
if isinstance(outcome, Deny):
raise PartDenied(name, outcome.reason, query.describe())
parts = (
list(outcome.parts[: limit - found])
if isinstance(outcome, Resolved)
else []
)
out[name] = parts
found += len(parts)
return out
[docs]
def distinct(self, query: CanonicalQuery, key: str) -> Mapping[str, Any]:
"""Values of `key` among matching parts, per provider that implements it.
Keyed by provider name rather than flattened, since which provider knows
a value is usually what the caller is trying to find out.
"""
out: dict[str, Any] = {}
for provider, name in zip(self, self.names(), strict=True):
values = provider.distinct(query, key)
if values is None: # this provider does not enumerate
continue
out[name] = list(values)
return out
# -- context ------------------------------------------------------------
[docs]
@classmethod
def get(cls) -> PartSource:
"""The active chain, or the one the environment supplies."""
from .policy import current_project, discovered_source
current = super().get()
# `is None`, not `or`: PartSource defines __len__, so an explicitly empty
# chain is falsy, and `with PartSource([]):` must mean "no providers here"
# rather than "fall back to the environment".
if current is not None:
return current
return discovered_source(current_project())
# `require()` is deliberately NOT overridden. The inherited one returns an
# `Instantiable` placeholder outside an active instantiation, which is what
# makes a class-body `PartSource.require()` work, and otherwise calls
# `cls.get()` above, which never returns None, so it never raises
# ContextMissingException. `NoChainConfigured` comes out of discovery instead.
