"""
Registration and chain discovery
================================
How a chain is found when the design did not hand one over, and how a string in a
chain becomes a provider instance.
Where a chain comes from, in precedence order:
1. a ``with PartSource([...])`` block
2. a ``parts`` attribute on the ``Design``
3. ``[tool.jitx.parts] chain`` in the current project's pyproject.toml
4. a chain published by an installed package, through the
``jitxlib-parts-policy`` entry point group
5. auto-discovery from the ``jitxlib-parts-provider`` group, opt-in only, behind
``discover = true``
Levels 1 and 2 come from the context stack, so :py:meth:`PartSource.get` handles
them. This module is levels 3 to 5, and a level is consulted only when every
earlier one declined. Configure none of them and `NoChainConfigured` says so,
listing the options, rather than a guess being made.
A provider package registers itself in the ``jitxlib-parts-provider`` group, which
is what lets a chain name it as a string. ``"csv:approved-parts.csv"`` splits on
the first colon: the name before it selects the entry point, the rest is the
provider's single constructor argument, and a path is taken relative to the
project. ``"ideal"`` takes no argument. A package publishing a whole chain instead
of a single provider registers in ``jitxlib-parts-policy``, which is level 4 and
the ordinary way an organization ships its own procurement policy.
:py:class:`Remapped` lives here rather than with the providers because it is chain
configuration rather than a provider of its own: it wraps one whose attribute
names differ from the ones a design uses, and it is written into the chain like
any other member.
"""
from __future__ import annotations
import functools
import inspect
from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from importlib.metadata import entry_points
from pathlib import Path
from typing import Any
# Aliased: this module already has a `discover = true` config key and a
# `discovered_source`, both about finding a *chain*. This one finds the project.
from jitx.project import JitxProject, NotAJitxProject, discover as discover_project
from .error import NoChainConfigured, ProviderContractError, ProviderUnavailable
from .provider import Outcome, PartSource, Provider, Resolved, Unsupported
from .query import CanonicalQuery, Predicate
from .record import JSON, PartIdentity, ResolvedPart
PROVIDER_GROUP = "jitxlib-parts-provider"
POLICY_GROUP = "jitxlib-parts-policy"
# --------------------------------------------------------------------------
# The string registry
# --------------------------------------------------------------------------
@functools.cache
def _provider_registry() -> Mapping[str, list[Any]]:
"""Entry points in the provider group, keyed by name.
A list per name rather than one entry, because two installed packages may
claim the same name and the error has to be able to list the candidates.
"""
out: dict[str, list[Any]] = {}
for ep in entry_points(group=PROVIDER_GROUP):
out.setdefault(ep.name, []).append(ep)
return out
[docs]
def resolve_provider_name(spec: str, root: Path | None = None) -> Provider:
"""Turn a chain string into a provider instance.
`"csv:approved-parts.csv"` splits on the first colon: `csv` names the entry
point, and the remainder is its single constructor argument. A bare `"ideal"`
constructs with no arguments.
`root` is passed on to providers that accept it, so a path written in project
config is relative to the project. A provider that takes no `root` is
constructed without one, which is why this inspects the signature rather than
requiring every provider to accept an argument it has no use for.
Ambiguity is an error naming the candidates, since the project's own config is
the documented way to disambiguate.
"""
name, _, argument = spec.partition(":")
candidates = _provider_registry().get(name, [])
if not candidates:
known = ", ".join(sorted(_provider_registry())) or "(none installed)"
# Not NoChainConfigured: a chain *is* configured, and one of its members
# cannot be reached. Passing this to NoChainConfigured also produced a
# mangled message, since that error formats its argument as a project root.
raise ProviderUnavailable(
f"chain names provider {name!r}, which is not installed. "
f"Available: {known}",
hint="Install the package that provides it, or correct the name in "
"this project's [tool.jitx.parts] chain",
)
if len(candidates) > 1:
where = ", ".join(str(getattr(c, "value", c)) for c in candidates)
raise ProviderContractError(
f"provider name {name!r} is claimed by more than one installed "
f"package ({where}); name the one you want in this project's "
f"[tool.jitx.parts] chain"
)
factory = candidates[0].load()
try:
if not argument:
return factory()
if root is not None and "root" in inspect.signature(factory).parameters:
return factory(argument, root=str(root))
return factory(argument)
except TypeError as exc:
# A chain entry is text somebody wrote in a config file, so a mismatch
# between it and the provider's signature is a configuration error, not a
# crash. The original is kept in the message, since it names the argument
# that was missing or surplus.
raise ProviderUnavailable(
f"chain entry {spec!r} does not match what provider {name!r} "
f"accepts: {exc}",
hint="A provider taking an argument is written `name:argument`, as "
"in `csv:parts/approved-parts.csv`; one taking none is just `name`",
) from exc
# --------------------------------------------------------------------------
# Project root and config
# --------------------------------------------------------------------------
[docs]
def current_project(start: Path | None = None) -> JitxProject | None:
"""The project `start` sits in, or None if it sits in none.
`start` defaults to the working directory. The returned project has already
read its pyproject.toml, so pass it on rather than looking it up again: both
`discovered_source` and `discovered_lock` take the project this returns.
None rather than a raise, because having no project is an ordinary state here
and not a failure: it is what a bare script or a notebook looks like, and it
turns the lockfile off and leaves only a published chain to find. Every caller
has a branch for it.
Cached per starting directory, so the file is read once rather than once per
part in the design. Reset with :py:func:`clear_project_cache`.
"""
# Keyed on the directory rather than cached outright, since one process may
# elaborate designs from two of them, and a chdir has to be noticed.
return _project_at(str(start if start is not None else Path.cwd()))
@functools.cache
def _project_at(start: str) -> JitxProject | None:
try:
return discover_project(start)
except NotAJitxProject:
return None
[docs]
def clear_project_cache() -> None:
"""Forget which project each directory belongs to, and its config with it."""
_project_at.cache_clear()
def _project_config(project: JitxProject) -> Mapping[str, Any]:
"""`[tool.jitx.parts]` from a project already read.
Empty means the project said nothing about parts, which every caller treats as
"this level declined".
"""
return project.tool("jitx", "parts") or {}
def _from_project_config(project: JitxProject) -> PartSource | None:
"""The chain named by `[tool.jitx.parts] chain`, if there is one.
Members are resolved with the project root in hand, so a path written in the
config is relative to the project rather than to whatever directory the build
happened to start in.
"""
chain = _project_config(project).get("chain")
if chain is None:
return None
return PartSource(
[
resolve_provider_name(m, project.root) if isinstance(m, str) else m
for m in chain
],
label="main",
)
def _from_published_chain() -> PartSource | None:
"""A chain published by an installed configuration package.
Two publishers is a hard error rather than a guess: picking one silently is
how a project ends up building against a chain nobody chose. The documented
remedy is naming the chain in this project's config, which is why this level
must not be reached when level 3 already answered.
"""
published = list(entry_points(group=POLICY_GROUP))
if not published:
return None
if len(published) > 1:
names = ", ".join(sorted(ep.name for ep in published))
raise ProviderContractError(
f"more than one installed package publishes a parts chain "
f"({names}); name the chain you want in this project's "
f"[tool.jitx.parts] chain"
)
value = published[0].load()
source = value() if callable(value) else value
if not isinstance(source, PartSource):
raise ProviderContractError(
f"{published[0].name} published {type(source).__name__}, "
f"expected a PartSource"
)
return source
def _from_discovery(project: JitxProject) -> PartSource | None:
"""Assemble a chain from whatever providers are installed. Opt-in only.
Ordered by `discovery_priority`, descending, then by name for stability. Only
zero-argument providers can be assembled this way, which is most of why this
level is a last resort rather than a feature.
"""
if not _project_config(project).get("discover", False):
return None
found: list[Provider] = []
for name in sorted(_provider_registry()):
candidates = _provider_registry()[name]
if len(candidates) != 1:
continue
factory = candidates[0].load()
try:
found.append(factory())
except TypeError:
# Needs an argument, so discovery cannot supply it.
continue
if not found:
return None
found.sort(key=lambda p: (-p.discovery_priority, p.name))
return PartSource(found, label="discovered")
_discovered_sources: dict[Path | None, PartSource] = {}
[docs]
def discovered_source(project: JitxProject | None) -> PartSource:
"""The chain implied by the installed environment, computed once per project.
Takes the project from :py:func:`current_project`, or `None` where there is no
project, such as a bare script or a notebook: only the published chain is
consulted then. Reset with :py:func:`clear_discovered_sources`.
"""
# Cached on the project root, not on the project object, for two reasons. One
# process may elaborate designs from two directories, so the root is the thing
# that distinguishes them; and `JitxProject` hashes by identity, so a fresh one
# per call would miss every time and grow the cache without bound.
#
# Cached at all so repeated calls return the same object rather than an equal
# one: a context's identity is load-bearing.
root = project.root if project is not None else None
# `not in`, never a falsiness test: `PartSource` defines `__len__`, so a
# legitimately empty chain is falsy and would be recomputed forever.
if root not in _discovered_sources:
_discovered_sources[root] = _discover(project)
return _discovered_sources[root]
[docs]
def clear_discovered_sources() -> None:
"""Drop every cached chain. For tests, and for a fresh run in one process."""
_discovered_sources.clear()
def _discover(project: JitxProject | None) -> PartSource:
"""Walk the precedence levels, stopping at the first that answers."""
# Callables, not values: a level must not be evaluated unless every earlier
# one declined. The two-publishers error above is only useful if reaching
# level 4 is what raises it, and level 5 imports every registered provider
# package, which must not happen when a chain is already configured.
levels = (
(
lambda: _from_project_config(project),
_from_published_chain,
lambda: _from_discovery(project),
)
if project is not None
else (_from_published_chain,)
)
# `is not None` at each step, never `or`: an explicit `chain = []` in project
# config is a legal way to say "no providers here", and it is falsy.
for level in levels:
candidate = level()
if candidate is not None:
return candidate
raise NoChainConfigured(project.root if project is not None else None)
# --------------------------------------------------------------------------
# Remapped: attribute vocabulary adaptation, from configuration
# --------------------------------------------------------------------------
[docs]
class Remapped(Provider):
"""Wrap a provider whose attribute names differ from the ones a design uses.
The map is canonical-to-provider: `{"case": "package_size"}` sends a query's
`case` filter as `package_size`, and rewrites the attribute back on the way
out so the design never sees the provider's spelling.
Part of the chain digest, since it is chain configuration living in the
project. So adding a remap re-resolves, exactly like changing a CSV path.
"""
# Not a dataclass: it normalizes a Mapping argument into a sorted tuple, so it
# needs a real __init__, and a frozen dataclass with one of those buys nothing.
# Equality and hashing come from Provider, over config().
inner: Provider
map: tuple[tuple[str, str], ...]
name: str
discovery_priority: int
def __init__(self, inner: Provider, map: Mapping[str, str] | None = None):
self.inner = inner
self.map = tuple(sorted((map or {}).items()))
self.name = inner.name
self.discovery_priority = inner.discovery_priority
def __repr__(self) -> str:
return f"Remapped({self.inner!r}, {dict(self.map)!r})"
[docs]
def config(self) -> dict[str, JSON]:
return {"remapped": self.inner.config(), "map": [list(p) for p in self.map]}
def _outbound(self) -> dict[str, str]:
return {v: k for k, v in self.map}
def _rewrite_query(self, query: CanonicalQuery) -> CanonicalQuery:
inbound = dict(self.map)
if not inbound:
return query
filters: dict[str, Predicate] = {
inbound.get(k, k): v for k, v in query.filters.items()
}
rank = tuple(
type(r)(inbound.get(r.key, r.key), r.direction) for r in query.rank
)
return CanonicalQuery(
category=query.category,
value_key=inbound.get(query.value_key or "", query.value_key),
filters=filters,
rank=rank,
limit=query.limit,
digest=query.digest,
)
def _rewrite_identity(self, identity: PartIdentity) -> PartIdentity:
outbound = self._outbound()
if not outbound:
return identity
from dataclasses import replace
attributes = {outbound.get(k, k): v for k, v in identity.attributes.items()}
return replace(identity, attributes=attributes)
def _rewrite_part(self, part: ResolvedPart) -> ResolvedPart:
rewritten = self._rewrite_identity(part.identity)
if rewritten is part.identity:
return part
return _RemappedPart(rewritten, part)
[docs]
def resolve(self, batch: Sequence[CanonicalQuery]) -> Sequence[Outcome]:
answers = list(self.inner.resolve([self._rewrite_query(q) for q in batch]))
out: list[Outcome] = []
for answer in answers:
if isinstance(answer, Resolved):
out.append(Resolved(tuple(self._rewrite_part(p) for p in answer.parts)))
elif isinstance(answer, Unsupported):
# A remapped provider declines in its own vocabulary, so the keys
# have to come back to canonical or the diagnostic names a key the
# user never wrote. Free-text reasons are left alone.
outbound = self._outbound()
out.append(
Unsupported(
tuple(outbound.get(k, k) for k in answer.keys), answer.reason
)
)
else:
out.append(answer)
return out
[docs]
def rehydrate(
self, identity: PartIdentity, payload: dict[str, JSON]
) -> ResolvedPart:
"""Re-apply the outbound map, or replay reports provider spellings."""
inbound = dict(self.map)
from dataclasses import replace
inner_identity = replace(
identity,
attributes={inbound.get(k, k): v for k, v in identity.attributes.items()},
)
return self._rewrite_part(self.inner.rehydrate(inner_identity, payload))
@dataclass(frozen=True)
class _RemappedPart:
"""A resolved part whose identity has been rewritten to canonical names."""
identity: PartIdentity
_inner: ResolvedPart = field(repr=False)
@property
def payload(self) -> dict[str, JSON]:
return self._inner.payload
def materialize(self, component: Any) -> None:
self._inner.materialize(component)
