"""Which flow an argument names: its id or its key, id first. The API takes ids and only ids. People type keys, and scripts written before ids existed pass keys, so the CLI, the SDK and the MCP tools answer the same question here, with the same words. Pure: the caller fetches what it may see with its own credentials and hands it over as ``(id, key)`` pairs. The SDK carries no engine code, so this lives with it, and the server's MCP tools import it from here. """ from __future__ import annotations import difflib from collections.abc import Iterable from dataclasses import dataclass from fluksio.sdk import SyncError __all__ = ["Resolved", "Unresolved", "resolve"] #: The explicit forms. A colon is legal in neither an id nor a key, so they #: cannot clash with a real value. ID = "id:" KEY = "key:" class Unresolved(SyncError): """Nothing, or more than one thing, answers to the argument.""" @dataclass(frozen=True) class Resolved: id: str #: Empty for an explicit id nobody lists any more. key: str #: Set when the argument was one flow's id and another's key, and the #: id was taken. Read-only callers show it; a change never gets this far. warning: str = "" def resolve( value: str, flows: Iterable[tuple[str, str]], *, mutation: bool = False ) -> Resolved: """The flow *value* names among *flows*. ``id:`` is taken as said, listed or not: history outlives its flow, and the engine answers for an id nobody holds. ``key:`` is the flow holding that key. A plain string is an id if one has it, else a key. A plain string that is one flow's id and another's key picks the id with a warning, or is refused when *mutation* says the call changes something. A key several flows answer to, which history can hold, is always refused. """ keys: dict[str, str] = {} for flow_id, key in flows: keys.setdefault(flow_id, key) if value.startswith(ID): named = value[len(ID) :] return Resolved(named, keys.get(named, "")) explicit = value.startswith(KEY) key = value[len(KEY) :] if explicit else value holders = [flow_id for flow_id, held in keys.items() if held == key] if not explicit and value in keys: others = [flow_id for flow_id in holders if flow_id != value] if not others: return Resolved(value, keys[value]) both = ( f"'{value}' is the id of '{keys[value]}' and the key of " f"'{', '.join(others)}'." ) if mutation: raise Unresolved(f"{both} Pass id:{value} or key:{value}.") return Resolved( value, keys[value], f"{both} Using the id. Pass key:{value} for the other." ) if len(holders) == 1: return Resolved(holders[0], key) if holders: raise Unresolved( f"Several flows have had the key '{key}': {', '.join(holders)}. " "Pass id:." ) close = difflib.get_close_matches(key, sorted(set(keys.values())), n=3) hint = f" Closest keys: {', '.join(close)}." if close else "" what = "key" if explicit else "id or key" raise Unresolved(f"No flow with the {what} '{key}'.{hint}")