Helpers
Field()
machinable.config.Field
python
Field(default: Any = PydanticUndefined, *, identifying: bool = True, **kwargs: Any)pydantic.Field with machinable extensions.
identifying=False marks the field as non-identifying: it is excluded from the config-identity hash (e.g. an environment-dependent storage URI). The flag is stored in a namespaced json_schema_extra slot so it never collides with user/JSON-schema keys, and is read back by identity_exclude.
import_ref()
machinable.config.import_ref
python
import_ref(ref: Any) -> AnyResolve and instantiate a reference (cls(**kwargs)); None -> None.
Use resolve_ref instead when you need to control construction.
resolve_ref()
machinable.config.resolve_ref
python
resolve_ref(ref: Any) -> tuple[typing.Any, dict] | NoneResolve a reference to (object, merged_kwargs).
A reference is a get(module, version) element, "pkg.obj" or ("pkg.obj", {...}, {...}), reusing machinable's own parser (extract) and deep merge (update_dict):
None -> None
"pkg.Foo" -> Foo, {}
("pkg.Foo", {"a": 1}) -> Foo, {"a": 1}
("pkg.Foo", {"a": 1}, {"a": 2}) -> Foo, {"a": 2} # deep, rightmost wins
predicate_from_manifest()
machinable.config.predicate_from_manifest
python
predicate_from_manifest(location: str, *keys: str, manifest: str = 'manifest.json') -> dictBuild a content predicate from keys in a blob's JSON manifest.
Convenience for on_compute_predicate in the volatile-location pattern: an interface excludes its environment-dependent *_uri from identity (via Field(..., identifying=False)) and re-identifies the underlying data here by a stable id stored in a manifest beside it. location may be a plain path or a scheme:// URI.
from_cli()
machinable.cli.from_cli
python
from_cli(args: list | None = None) -> 'VersionType'Parse CLI-style arguments into a compact version list.
Source
src/machinable/config.py
py
"""Configuration: pydantic Config handling, canonical identity, references."""
__all__ = [
"to_dict",
"Field",
"identity_exclude",
"canonical",
"canonical_identity_key",
"predicate_from_manifest",
"import_object_by_path",
"resolve_ref",
"import_ref",
]
import collections.abc
import importlib
import re
import warnings
from inspect import isclass
from typing import TYPE_CHECKING, Any, cast, get_args, get_origin
import omegaconf
from pydantic import BaseModel, ValidationError
from pydantic import Field as PydanticField
from pydantic_core import PydanticUndefined
from machinable.errors import ConfigurationError
if TYPE_CHECKING:
from machinable.interface import Interface
class _ModelPrototype(BaseModel):
pass
def _nested_model(annotation: Any) -> "type[BaseModel] | None":
"""The single BaseModel inside an annotation, or ``None``.
Resolves a direct model and a model behind ``Optional``/unions; a union of
several models is ambiguous and yields ``None``.
"""
if isclass(annotation) and issubclass(annotation, BaseModel):
return annotation
models = {
found
for arg in get_args(annotation)
if (found := _nested_model(arg)) is not None
}
return models.pop() if len(models) == 1 else None
def check_unknown_keys(
config: collections.abc.Mapping, model: type[BaseModel], _path: str = ""
) -> None:
"""Raise on config keys that the Config (or a nested model) does not declare.
Pydantic ignores extra keys by default, so a typo in an update would
validate silently against the defaults and, since config is identity,
dedup onto the wrong record. A model that sets ``extra`` in its
``model_config`` (directly or inherited) keeps its own choice at that
level; nested models are checked through ``Optional``/single-model unions
and ``list``/``dict`` containers.
"""
handles_extras = "extra" in model.model_config
for key, value in config.items():
field = model.model_fields.get(key)
if field is None:
if handles_extras:
continue
declared = ", ".join(model.model_fields) or "<none>"
raise ConfigurationError(
f"Unknown config key '{_path}{key}' "
f"(declared fields: {declared}). Fix the name, or set "
f'model_config = ConfigDict(extra="allow") on the Config '
f"to accept free-form keys."
)
_check_annotation(value, field.annotation, f"{_path}{key}")
def _check_annotation(value: Any, annotation: Any, path: str) -> None:
origin = get_origin(annotation)
args = get_args(annotation)
if origin in (list, tuple, set, frozenset) and isinstance(value, list | tuple):
if args:
for i, item in enumerate(value):
_check_annotation(item, args[0], f"{path}[{i}]")
return
if (
origin in (dict, collections.abc.Mapping)
and len(args) == 2
and isinstance(value, collections.abc.Mapping)
):
for k, v in value.items():
_check_annotation(v, args[1], f"{path}.{k}")
return
nested = _nested_model(annotation)
if nested is not None and isinstance(value, collections.abc.Mapping):
check_unknown_keys(value, nested, f"{path}.")
def from_interface(
interface: "Interface | type[Interface]",
) -> "tuple[dict, type[BaseModel] | None]":
"""The default config dict and Config model class of an interface."""
if not isclass(interface):
interface = interface.__class__
if not hasattr(interface, "Config"):
return {}, None
config = getattr(interface, "Config")
# pydantic model, the only supported Config form
if isinstance(config, type(_ModelPrototype)):
with warnings.catch_warnings(record=True): # ignore pydantic warnings
try:
return config().model_dump(), config
except ValidationError:
# required field(s) without a default: pydantic forbids the bare
# ``config()`` instantiation, so build the default map per-field.
return _model_defaults(config), config
raise ConfigurationError(
f"Interface Config must subclass pydantic.BaseModel, but "
f"{interface.__name__}.Config is a {type(config).__name__}. Plain dict and "
f"dataclass/bare-class Config are no longer supported; define "
f"`class Config(pydantic.BaseModel): ...`."
)
def _model_defaults(model: type[BaseModel]) -> dict:
"""Default config for a pydantic model, tolerant of required fields.
Mirrors ``model().model_dump()`` but never instantiates the model, so a field
without a default (``PydanticUndefined``) is omitted rather than raising. Such
required fields are simply absent from the ``_default_`` layer, which is exactly
right for identity: with no default to compare against they are never
default-stripped, hence always part of the canonical form. Nested models are
recursed into so their defaults are represented too.
"""
out: dict = {}
for name, field in model.model_fields.items():
default = field.get_default(call_default_factory=True)
if default is PydanticUndefined:
continue # required: omit; default-strip keeps the supplied value whole
if isinstance(default, BaseModel):
out[name] = default.model_dump()
else:
out[name] = default
return out
def match_method(definition: str) -> tuple[str, str] | None:
"""Parse a ``name(args)`` reference into ``(name, args)``, or ``None``.
``args`` may contain balanced nested parentheses (``fit(bounds=(0, 1))``).
Expression strings like ``"foo(1) + bar(2)"`` do not match, since their
final parenthesis does not close the opening one; consumers additionally
verify that ``name`` refers to a real method before acting on a match.
"""
fn_match = re.match(r"^(?P<method>\w+)\s?\((?P<args>.*)\)$", definition)
if fn_match is None:
return None
args = fn_match.groupdict()["args"]
depth = 0
for char in args:
if char == "(":
depth += 1
elif char == ")":
depth -= 1
if depth < 0: # closes the outer call early: not a single call
return None
if depth != 0:
return None
return fn_match.groupdict()["method"], args
def _has_config_method(interface: "Interface | None", function: str) -> bool:
"""True when ``function(...)`` refers to a real config method.
Checks the interface and the project provider. Without an interface
context (legacy callers) assume yes, preserving prior behaviour. This
stops arbitrary strings that merely contain parentheses (e.g. a label
``"foo (bar)"``) from being misread as config methods.
"""
if interface is None:
return True
method = "config_" + function
if hasattr(interface, method):
return True
try:
from machinable.project import Project
return hasattr(Project.get().provider(), method)
except Exception:
return False
def rewrite_config_methods(
config: collections.abc.Mapping | str | list | tuple,
interface: "Interface | None" = None,
) -> Any:
"""Rewrite config-method references into omegaconf resolver calls."""
if isinstance(config, list):
return [rewrite_config_methods(v, interface) for v in config]
if isinstance(config, tuple):
return (rewrite_config_methods(v, interface) for v in config)
if isinstance(config, collections.abc.Mapping):
items = cast("collections.abc.Mapping[Any, Any]", config).items()
return {k: rewrite_config_methods(v, interface) for k, v in items}
if isinstance(config, str):
matched = match_method(config)
if matched is not None and _has_config_method(interface, matched[0]):
function, args = matched
return '${config_method:"' + function + '","' + args + '"}'
return config
def to_dict(dict_like):
"""Convert omegaconf containers (recursively) into plain dicts."""
if isinstance(dict_like, omegaconf.DictConfig | omegaconf.ListConfig):
return omegaconf.OmegaConf.to_container(dict_like)
if isinstance(dict_like, list | tuple):
return dict_like.__class__([k for k in dict_like])
if not isinstance(dict_like, collections.abc.Mapping):
return dict_like
return {k: to_dict(v) for k, v in dict_like.items()}
def Field(default: Any = PydanticUndefined, *, identifying: bool = True, **kwargs: Any):
"""``pydantic.Field`` with machinable extensions.
``identifying=False`` marks the field as *non-identifying*: it is excluded from
the config-identity hash (e.g. an environment-dependent storage URI). The flag
is stored in a namespaced ``json_schema_extra`` slot so it never collides with
user/JSON-schema keys, and is read back by :func:`identity_exclude`.
"""
if identifying is False:
extra = dict(kwargs.pop("json_schema_extra", None) or {})
namespaced = dict(extra.get("machinable", {}))
namespaced["identifying"] = False
extra["machinable"] = namespaced
kwargs["json_schema_extra"] = extra
return PydanticField(default, **kwargs)
def identity_exclude(config_cls: Any, _prefix: str = "") -> set[str]:
"""Dotted field paths a pydantic ``Config`` marks non-identifying via :func:`Field`.
Recurses into nested models (direct or behind ``Optional``), producing
paths like ``data.uri``. Models inside containers (``list[Sub]``) are not
reached since exclusion there has no addressable path.
"""
out: set[str] = set()
for name, field in getattr(config_cls, "model_fields", {}).items():
extra = getattr(field, "json_schema_extra", None)
if isinstance(extra, collections.abc.Mapping):
namespaced = extra.get("machinable")
if (
isinstance(namespaced, collections.abc.Mapping)
and namespaced.get("identifying") is False
):
# the whole subtree leaves identity; no need to recurse
out.add(_prefix + name)
continue
nested = _nested_model(field.annotation)
if nested is not None:
out |= identity_exclude(nested, f"{_prefix}{name}.")
return out
def canonical(
resolved: collections.abc.Mapping,
default: collections.abc.Mapping | None = None,
exclude: collections.abc.Collection[str] = (),
) -> dict:
"""Canonical (identity) form of a *resolved* config mapping.
Recursively strips keys that equal their default (compared against ``default``,
machinable's ``_default_`` layer) and drops non-identifying ``exclude`` fields
(dotted paths reaching into nested mappings, e.g. ``data.uri``). Keys absent
from ``default`` (free-form deviations and required fields) are kept
verbatim. Numerics are **not** folded: pydantic coercion has already aligned
types upstream, so a plain ``==`` against the coerced default is correct, and
free-form container contents are hashed as-is.
"""
default = default or {}
out: dict = {}
for key, value in resolved.items():
if key in exclude:
continue
nested_exclude = {
path[len(key) + 1 :] for path in exclude if path.startswith(f"{key}.")
}
if key in default:
dv = default[key]
if isinstance(value, collections.abc.Mapping) and isinstance(
dv, collections.abc.Mapping
):
sub = canonical(value, dv, nested_exclude)
if sub: # whole nested dict at default → stripped
out[key] = sub
continue
if value == dv:
continue
elif nested_exclude and isinstance(value, collections.abc.Mapping):
sub = canonical(value, {}, nested_exclude)
if sub:
out[key] = sub
continue
out[key] = value
return out
def canonical_identity_key(
module: str | None,
resolved: collections.abc.Mapping,
default: collections.abc.Mapping | None = None,
exclude: collections.abc.Collection[str] = (),
) -> str:
"""Content-addressed config identity: ``hash(module + canonical(config))``.
The single source of truth for ``identity_key`` so the live (materialize) and
reindex sites cannot drift. Default-strip needs only the resolved/`_default_`
layers (always present in ``model.json``); ``exclude`` (``identifying=False``
fields) needs the Config class and is supplied where available.
"""
from machinable.utils import object_hash
return object_hash(
{"module": module, "config": canonical(resolved, default, exclude)}
)[:24]
def predicate_from_manifest(
location: str,
*keys: str,
manifest: str = "manifest.json",
) -> dict:
"""Build a content predicate from keys in a blob's JSON manifest.
Convenience for ``on_compute_predicate`` in the volatile-location pattern: an
interface excludes its environment-dependent ``*_uri`` from identity (via
``Field(..., identifying=False)``) and re-identifies the underlying data here by
a stable id stored in a manifest beside it. ``location`` may be a plain path or a
``scheme://`` URI.
"""
from machinable.utils import load_file, uri_to_path
base = uri_to_path(location) if "://" in str(location) else location
data = load_file([base, manifest], None)
if not isinstance(data, collections.abc.Mapping):
raise FileNotFoundError(f"No manifest {manifest!r} found at {base!r}")
return {key: data[key] for key in keys}
def import_object_by_path(path: str) -> Any:
"""Import an object by a dotted or ``module:attr`` path.
Resolves through :func:`importlib.import_module` so a given path maps to the
single ``sys.modules``-cached object, avoiding the duplicate-class import
boundary (two loads of the same module under different names → distinct class
objects). Use canonically importable paths.
"""
if ":" in path:
module_path, _, attr = path.partition(":")
else:
module_path, _, attr = path.rpartition(".")
if not module_path or not attr:
raise ValueError(f"Invalid import path: {path!r}")
obj: Any = importlib.import_module(module_path)
for part in attr.split("."):
obj = getattr(obj, part)
return obj
def resolve_ref(ref: Any) -> tuple[Any, dict] | None:
"""Resolve a reference to ``(object, merged_kwargs)``.
A reference is a ``get(module, version)`` element, ``"pkg.obj"`` or
``("pkg.obj", {...}, {...})``, reusing machinable's own parser (``extract``)
and deep merge (``update_dict``):
None -> None
"pkg.Foo" -> Foo, {}
("pkg.Foo", {"a": 1}) -> Foo, {"a": 1}
("pkg.Foo", {"a": 1}, {"a": 2}) -> Foo, {"a": 2} # deep, rightmost wins
"""
if ref is None:
return None
from machinable.interface import extract # lazy: interface imports config
from machinable.utils import update_dict
module, versions = extract(ref)
if module is None:
raise ValueError(f"Invalid reference: {ref!r}")
obj = import_object_by_path(module)
kwargs: dict = {}
for version in versions or []:
kwargs = update_dict(kwargs, version)
return obj, to_dict(kwargs)
def import_ref(ref: Any) -> Any:
"""Resolve and instantiate a reference (``cls(**kwargs)``); ``None`` -> ``None``.
Use :func:`resolve_ref` instead when you need to control construction.
"""
resolved = resolve_ref(ref)
if resolved is None:
return None
obj, kwargs = resolved
return obj(**kwargs)src/machinable/cli.py
py
"""The ``machinable`` command-line interface."""
import os
import sys
from typing import TYPE_CHECKING, cast
from omegaconf import OmegaConf
if TYPE_CHECKING:
from machinable.types import VersionType
def parse(args: list) -> tuple:
"""Parse CLI arguments into elements, kwargs, and method calls."""
kwargs = []
methods = []
elements = []
dotlist = []
version = []
def _parse_dotlist():
if len(dotlist) == 0:
return
_ver = {}
container = cast(dict, OmegaConf.to_container(OmegaConf.from_dotlist(dotlist)))
for k, v in container.items():
key = str(k)
if key.startswith("**"):
kwargs[-1][key[2:]] = v
else:
_ver[key] = v
version.append(_ver)
def _push():
if len(version) == 0:
return
if len(elements) > 0:
elements[-1].extend(version)
else:
elements.append(version)
for arg in args:
if arg.startswith("~"):
# version
_parse_dotlist()
dotlist = []
version.append(arg)
elif arg.startswith("--"):
# method
methods.append((len(elements), arg[2:]))
elif "=" in arg:
# dotlist
dotlist.append(arg)
else:
# module
_parse_dotlist()
_push()
dotlist = []
version = []
# auto-complete `.project` -> `interface.project`
if arg.startswith("."):
arg = "interface" + arg
elements.append([arg])
kwargs.append({})
_parse_dotlist()
_push()
return elements, kwargs, methods
def from_cli(args: list | None = None) -> "VersionType":
"""Parse CLI-style arguments into a compact version list."""
if args is None:
args = sys.argv[1:]
elements, _, _ = parse(args)
return sum(elements, [])
def _run_mcp(args: list) -> int:
"""Launch the research MCP server (stdio by default).
``machinable mcp [--project DIR] [--token T] [--read-only]
[--http --host H --port P]``
"""
project = os.getcwd()
token = None
read_only = False
transport = "stdio"
host, port = "127.0.0.1", 8765
it = iter(args)
for arg in it:
if arg == "--project":
project = next(it)
elif arg == "--token":
token = next(it)
elif arg == "--read-only":
read_only = True
elif arg == "--http":
transport = "http"
elif arg == "--host":
host = next(it)
elif arg == "--port":
port = int(next(it))
from machinable.mcp.server import create_server
server = create_server(project, token=token, read_only=read_only)
if transport == "http":
server.run(transport="http", host=host, port=port)
else:
server.run()
return 0
def main(args: list | None = None):
"""Entry point of the ``machinable`` command."""
import machinable
if args is None:
args = sys.argv[1:]
if len(args) == 0:
print("\nhelp")
print("\nversion")
print("\nget")
return 0
action, args = args[0], args[1:]
if action == "help":
h = "get"
if len(args) > 0:
h = args[0]
if h == "get":
print("\nmachinable get [interface_module...] [version...] --method")
print("\nExample:")
print("\tmachinable get my_interface ~ver arg=1 nested.arg=2 --launch\n")
print("Methods accept arguments like versions do: --summary(top=3)")
print("(quote the token if your shell parses parentheses)\n")
return 0
elif h == "version":
print("\nmachinable version")
return 0
elif h == "fetch":
print("\nmachinable fetch [module ...] [--project DIR]")
print(
"\nDownload declared remotes into interface/remotes/ without "
"importing them,\nso the code can be inspected before it "
"executes. Without arguments, fetches\nevery declared remote."
)
return 0
else:
print("Unrecognized option")
return 128
if action == "version":
version = machinable.get_version()
print(version)
return 0
if action == "mcp":
return _run_mcp(args)
if action == "dispatch":
# `machinable dispatch [--foreground|--detach|--prepare] [--wait]
# [--with-metadata] [--project DIR]`, the serverless mirror of
# POST /v1/executions; DispatchRequest JSON on stdin (see
# machinable.dispatch).
from machinable.dispatch import main as dispatch_main
return dispatch_main(args)
if action == "migrate":
# `machinable migrate [storage_directory]`: one-shot in-place upgrade
# of a store to format v1 (id.json + updated_at; hidden → overlay).
from machinable.migrate import migrate_store
summary = migrate_store(args[0] if args else "./storage")
print(
f"Migrated {summary['directory']}: "
f"{summary['upgraded']} upgraded, "
f"{summary['already_v1']} already v1, "
f"{summary['reindexed']} reindexed, "
f"{summary['hidden']} hidden flags moved to overlay"
)
return 0
if action == "fetch":
# `machinable fetch [module ...] [--project DIR]`: download declared
# remotes into interface/remotes/ WITHOUT importing them, so the code
# can be inspected before it is ever executed (remote modules run
# arbitrary code on import).
project_dir = os.getcwd()
modules = []
it = iter(args)
for arg in it:
if arg == "--project":
project_dir = next(it)
else:
modules.append(arg)
sys.path.append(project_dir)
from machinable.project import Project
with Project(project_dir) as project:
provider = project.provider()
declared = provider.on_resolve_remotes() or {}
if not modules:
modules = list(declared)
if not modules:
print("No remotes declared (see Project.on_resolve_remotes)")
return 0
status = 0
for module in modules:
filename = provider.fetch_remote(module)
if filename is None:
print(f"{module}: not a declared remote")
status = 1
else:
print(f"{module} -> {filename}")
return status
if action == "touch":
# `machinable touch <record_id>`: declare an out-of-band change so
# updated_at (and the index) reflect it.
if not args:
print("Usage: machinable touch <record_id>")
return 128
sys.path.append(os.getcwd())
from machinable.interface import Interface
from machinable.project import Project
with Project(os.getcwd()):
interface = Interface.find_by_id(args[0], fetch=False)
if interface is None:
print(f"No record '{args[0]}' in the index")
return 1
interface.touch()
print(f"Touched {interface.uuid}")
return 0
if action.startswith("get"):
sys.path.append(os.getcwd())
get = machinable.get
if action != "get":
get = getattr(get, action.split(".")[-1])
elements, kwargs, methods = parse(args)
contexts = []
target_interface = None
for i, (module, *version) in enumerate(elements):
interface = get(module, version, **kwargs[i])
if i == len(elements) - 1:
target_interface = interface
else:
contexts.append(interface.__enter__())
if target_interface is None:
raise ValueError("You have to provide at least one interface")
from machinable.config import match_method
for i, method in methods:
# `--method(a=1)` passes arguments, like `~version(a=1)`
name, call_args = method, None
if method.endswith(")"):
matched = match_method(method)
if matched is None:
print(f"Invalid method: --{method}")
return 128
name, call_args = matched
# check if cli_{name} exists before falling back on {name}
target = getattr(
target_interface,
f"cli_{name}",
getattr(target_interface, name),
)
if callable(target):
try:
result = eval( # pylint: disable=eval-used
"__target__(" + (call_args or "") + ")",
{"__target__": target},
)
except (SyntaxError, TypeError) as _ex:
print(f"Invalid arguments in --{method}: {_ex}")
return 128
# print returned values; fluent methods returning the
# interface itself (e.g. --launch) stay quiet
if result is not None and result is not target_interface:
print(result)
else:
print(target)
for context in reversed(contexts):
context.__exit__()
return 0
print("Invalid argument")
return 128
if __name__ == "__main__":
raise SystemExit(main())