Project
machinable.project.Project (inherits Interface) · source
The project directory a session binds to.
Also the root entry every other record scopes under.
Config
| Field | Type | Default |
|---|---|---|
directory | str | None | computed |
add_to_path()
python
add_to_path(self) -> NonePut the project directory on sys.path.
claims
The verified claims of the API request served in this project.
Read-only; set when the API server admitted the request with an OIDC token (the identity claim is also username), empty otherwise.
compute_predicate()
python
compute_predicate(self) -> dictThe root predicate.
The project's directory name only; the root never absorbs ambient scopes.
compute_record_id()
python
compute_record_id(self) -> strDeterministic root record id, derived from the identity triple.
Every index mints the same root uuid for the same project, so cross-peer id.json parent references resolve on ingest and the interface dedup triple (parent, identity, predicate) matches across machines, where random per-index root ids would orphan every shared record.
ensure_project_root()
python
ensure_project_root(self)Lazily materialize this Project as its own first index entry.
The Project is a first-class interface whose directory is the project directory itself; every interface in the project is scoped under this root entry. Idempotent, and resilient to a rebuilt/wiped index (the cached entry is re-validated and re-created if missing).
exists()
python
exists(self) -> boolTrue when the project directory exists.
fetch_remote()
python
fetch_remote(self, module: str) -> str | NoneFetch a declared remote module into interface/remotes/, no import.
Recursively fetches declared dependencies and returns the local file path, or None when module is not a declared remote. Since remote modules execute on import, fetching first (machinable fetch) lets you inspect the code before anything runs.
get_host_info()
python
get_host_info(self) -> dictReturned dictionary will be recorded as host information.
get_root()
python
get_root(self) -> 'Project'The root project of a vendor chain.
get_vendors()
python
get_vendors(self) -> list[str]Names of vendored dependencies under vendor/.
index
The connected (or default) Index.
interface()
python
interface(self, module: 'str | Interface | type[Interface]', version: str | dict | None | collections.abc.Sequence[str | dict | None] = None, base_class: Any = None, **constructor_kwargs) -> machinable.interface.InterfaceResolve and instantiate an interface module from this project.
is_root()
python
is_root(self) -> boolTrue when this project has no parent project.
module
The provider's module path.
module_schema()
python
module_schema(self, module: str)The static config schema for one discovered module.
modules()
python
modules(self) -> listStatically discovered interface modules.
name()
python
name(self) -> strThe project directory's name (its portable discriminator).
on_compute_predicate()
python
on_compute_predicate(self) -> dictThe directory name: the portable discriminator between projects.
on_resolve_discovery()
python
on_resolve_discovery(self) -> listDiscovery schemes consulted when enumerating this project's modules.
Returns importable Discovery elements (a module string, a module:Class reference, or a class), resolved in order and composed first-wins. The default is the built-in static AST discovery.
on_resolve_interface()
python
on_resolve_interface(self, module: 'str | Interface | type[Interface]') -> 'tuple[str | Interface | type[Interface], type[Interface] | None]'Override interface resolution.
Return altered module and/or resolved Interface class to be used instead.
on_resolve_manifests()
python
on_resolve_manifests(self) -> list[str]Code-provenance tools to capture for each run.
Returns importable Manifest modules; the default is the built-in git manifest. Override in your project to add or replace tools (a dependency lock, a container digest); each is an ordinary machinable.manifest.Manifest subclass, resolved like any interface.
on_resolve_remotes()
python
on_resolve_remotes(self) -> 'dict[str, str | list[str]]'Event triggered during remote resolution.
Return a dictionary of module names and their remote source (or a [source, *dependency_modules] list).
on_resolve_vendor()
python
on_resolve_vendor(self, name: str, source: str | None, target: str) -> bool | NoneEvent triggered when vendor is resolved.
Arguments
name: The name of the vendor source: The source configuration target: The target directory (may or may not exists yet)
Return False to prevent the default automatic resolution
path()
python
path(self, *append: str) -> strAbsolute path inside the project directory.
provider()
python
provider(self, reload: str | bool = False) -> 'Project'Resolves and returns the provider instance.
reindex()
python
reindex(self, storage=None, *, relations: bool = True) -> list[str](Re)build this project's Index by scanning a Storage for.
.machinable markers. Defaults to the project's connected Storage (a directory under ./storage). Returns the ingested uuids.
resolve_remotes()
python
resolve_remotes(self, module: str) -> machinable.interface.Interface | NoneFetch and import a remote module declared by on_resolve_remotes.
serialize()
python
serialize(self) -> dictThe project as a plain dict.
storage
The connected (or default) Storage.
unserialize()
python
unserialize(cls, serialized: dict) -> 'Project'Rebuild a project from serialize() output.
username
The user attributed as creator of interfaces in this project.
Defaults to the OS user.
Source
src/machinable/project.py
py
"""Projects: the directory a session binds to, and vendor/remote resolution."""
import getpass
import importlib
import os
import platform
import shutil
import socket
import subprocess
import sys
import urllib.request
from collections.abc import Mapping
from types import MappingProxyType
from typing import Any, cast
from pydantic import BaseModel, field_validator
import machinable
from machinable import schema
from machinable.config import Field
from machinable.errors import ConfigurationError
from machinable.interface import (
Interface,
extend,
get_inherits,
instantiate,
)
from machinable.types import VersionType
from machinable.utils import (
find_subclass_in_module,
import_from_directory,
is_directory_version,
)
def _symlink_or_copy(source: str, target: str) -> None:
"""Symlink ``source`` -> ``target``, or copy where symlinks are forbidden.
Windows without admin/Developer Mode raises ``OSError`` on symlink.
"""
try:
os.symlink(source, target, target_is_directory=True)
except OSError:
if os.path.isdir(source):
shutil.copytree(source, target, symlinks=True)
else:
shutil.copy2(source, target)
def fetch_link(source, target):
"""Vendor fetcher: symlink (or copy) ``source`` to ``target``."""
_symlink_or_copy(source, target)
return True
def fetch_directory(source, target):
"""Vendor fetcher: link a local directory."""
if not os.path.isdir(source):
raise ValueError(f"{source} is not a directory")
# we use symlink rather than copy for performance and consistency
fetch_link(source, target)
return True
def fetch_git(source: str, target: str) -> bool:
"""Vendor fetcher: clone a git repository."""
name, directory = os.path.split(target)
command = ["git", "clone", source, name]
process = subprocess.Popen(command, cwd=directory)
process.communicate()
if process.returncode != 0:
return False
return True
def fetch_vendor(source, target):
"""Fetch a vendor from its source spec (``git+``, ``link+``, or directory)."""
if source is None:
return False
# git
if source.startswith("git+"):
return fetch_git(source[4:], target)
# symlink
if source.startswith("link+"):
return fetch_link(source[5:], target)
# default: directory
return fetch_directory(source, target)
def fetch_vendors(project: "Project"):
"""Fetch every declared vendor into the project."""
top_level_vendor = project.get_root().path("vendor")
os.makedirs(top_level_vendor, exist_ok=True)
vendors = []
for dependency in project.get_vendors():
if isinstance(dependency, str):
dependency = {dependency: None}
for name, args in dependency.items():
vendors.append(name)
# source config
if not isinstance(args, str):
args = None
# give local .machinablerc priority over project config
source = args
if isinstance(source, str):
source = os.path.expanduser(source)
# local target folder
os.makedirs(project.path("vendor"), exist_ok=True)
target = project.path("vendor", name)
# protect against invalid symlinks
if os.path.islink(target) and not os.path.exists(target):
os.unlink(target)
# skip if already existing
if os.path.exists(target):
break
# top-level target folder
top_level = os.path.join(top_level_vendor, name)
# protect against invalid symlinks
if os.path.islink(top_level) and not os.path.exists(top_level):
os.unlink(top_level)
if project.on_resolve_vendor(name, source, top_level) is not False:
# fetch import to the top-level if it does not exist
if not os.path.exists(top_level):
print(f"Fetching '+.{name}' to {top_level}")
if not fetch_vendor(source, top_level):
raise FileNotFoundError(
f"Could not fetch '+.{name}'."
f" Please place it into {top_level}"
)
# symlink from top-level to local target if not identical
if top_level != target:
_symlink_or_copy(top_level, target)
break
return vendors
def import_interface(
directory: str, module: str, base_class: Any = None
) -> type[Interface]:
"""Import ``module`` from ``directory`` and return its interface class."""
if base_class is None:
base_class = Interface
# import project-relative, falling back to a global import
resolved = import_from_directory(module, directory, or_fail=False)
if resolved is None:
try:
resolved = importlib.import_module(module)
except ModuleNotFoundError as _e:
raise ModuleNotFoundError(
f"Could not find module '{module}' in project {directory}"
) from _e
interface_class = find_subclass_in_module(resolved, base_class)
if interface_class is None:
raise ConfigurationError(
f"Could not find an interface inheriting from the"
f" {base_class.__name__} base class. "
f"Is it correctly defined in {resolved.__name__}?"
)
return interface_class
class Project(Interface):
"""The project directory a session binds to.
Also the root entry every other record scopes under.
"""
kind = "Project"
_ambient = True
class Config(BaseModel):
# Non-identifying (design/format.md: identity lives in files, never
# paths): the project's filesystem location must not enter its
# identity key, or one project splits into a different root per
# checkout location. The portable discriminator between projects is
# the directory *name*, contributed via `on_compute_predicate`.
"""``directory`` is non-identifying: location never enters identity."""
directory: str | None = Field(
default_factory=lambda: os.getcwd(), identifying=False
)
@field_validator("directory")
def normalize_directory(cls, v):
"""Normalize the directory to an absolute path."""
return os.path.normpath(os.path.abspath(os.path.expanduser(v)))
def __init__(
self,
version: VersionType = None,
username: str | None = None,
claims: Mapping[str, Any] | None = None,
):
if is_directory_version(version):
version = {"directory": version}
super().__init__(version=version)
self.__model__ = schema.Project(
kind=self.kind,
version=self.__model__.version,
inherits=get_inherits(self),
)
self._parent: Project | None = None
self._provider: str = "interface/project"
self._resolved_provider: Project | None = None
self._root_entry = None
self._username = username
# verified token claims of the API request this project serves (OIDC)
self._claims = MappingProxyType(dict(claims)) if claims else None
@property
def username(self) -> str:
"""The user attributed as creator of interfaces in this project.
Defaults to the OS user.
"""
return self._username or getpass.getuser()
@property
def claims(self) -> Mapping[str, Any]:
"""The verified claims of the API request served in this project.
Read-only; set when the API server admitted the request with an OIDC
token (the identity claim is also :attr:`username`), empty otherwise.
"""
return self._claims if self._claims is not None else MappingProxyType({})
def __enter__(self):
super().__enter__()
self._root_entry = None
return self
@property
def index(self):
"""The connected (or default) Index."""
from machinable.index import Index
return Index.get()
@property
def storage(self):
"""The connected (or default) Storage."""
from machinable.storage import Storage
return Storage.get()
def reindex(self, storage=None, *, relations: bool = True) -> list[str]:
"""(Re)build this project's Index by scanning a Storage for.
``.machinable`` markers. Defaults to the project's connected Storage
(a directory under ``./storage``). Returns the ingested uuids.
"""
from machinable.index import Index
return Index.get().reindex(storage, relations=relations)
def ensure_project_root(self):
"""Lazily materialize this Project as its own first index entry.
The Project is a first-class interface whose directory is the project
directory itself; every interface in the project is scoped under this
root entry. Idempotent, and resilient to a rebuilt/wiped index (the
cached entry is re-validated and re-created if missing).
"""
from machinable.index import Index
index = Index.get()
if self._root_entry is not None:
if index.get_by_id(self._root_entry.record_id) is None:
self._root_entry = None
if self._root_entry is None:
self._root_entry = index.materialize_from(self).entry
return self._root_entry
def provider(self, reload: str | bool = False) -> "Project":
"""Resolves and returns the provider instance."""
if isinstance(reload, str):
self._provider = reload
self._resolved_provider = None
if self._resolved_provider is None or reload:
if isinstance(self._provider, str):
# without an interface/project.py, fall back to the connected
# class itself so a directly connected Project subclass keeps
# its hook overrides
self._resolved_provider = find_subclass_in_module(
module=import_from_directory(self._provider, self.config.directory),
base_class=Project,
default=type(self),
)(version=self.__model__.version)
else:
self._resolved_provider = Project(version=self.__model__.version)
return self._resolved_provider
@property
def module(self) -> str | None:
"""The provider's module path."""
if self.provider().__module__ != "machinable.project":
return self._provider.replace("/", ".")
return self.provider().__module__
def add_to_path(self) -> None:
"""Put the project directory on ``sys.path``."""
if (
os.path.exists(self.config.directory)
and self.config.directory not in sys.path
):
if self.is_root():
sys.path.insert(0, self.config.directory)
else:
sys.path.append(self.config.directory)
def name(self) -> str:
"""The project directory's name (its portable discriminator)."""
return os.path.basename(self.config.directory)
def on_compute_predicate(self) -> dict:
# The portable discriminator between projects (identity excludes the
# location, so something must distinguish two different projects in
# one index): the directory *name*. Two checkouts of "my-study" on
# different machines are the same project; by design, this is what
# makes collaborator ingest land under the same root.
"""The directory name: the portable discriminator between projects."""
return {"name": self.name()}
def compute_predicate(self) -> dict:
# Unlike work interfaces, the project root must NOT absorb ambient
# Scope predicates: ensure_project_root() runs inside whatever context
# happened to trigger the first materialization, and a scope-polluted
# root predicate would mint a different root per scope (with the
# deterministic record id below, a loud IndexCollision; before it, a
# silent duplicate root).
"""The root predicate.
The project's directory name only; the root never absorbs ambient
scopes.
"""
return self.on_compute_predicate()
def compute_record_id(self) -> str:
"""Deterministic root record id, derived from the identity triple.
Every index mints the *same* root uuid for the same project, so
cross-peer ``id.json`` parent references resolve on ingest and the
interface dedup triple (parent, identity, predicate) matches across
machines, where random per-index root ids would orphan every shared
record.
"""
from machinable.utils import object_hash, sanitize_filename
identity = sanitize_filename(self.catalog_identity_key(), max_len=128)
return (
"P"
+ object_hash({"identity": identity, "predicate": self.predicate_key})[:11]
)
def path(self, *append: str) -> str:
"""Absolute path inside the project directory."""
return os.path.join(self.config.directory, *append)
def is_root(self) -> bool:
"""True when this project has no parent project."""
return self._parent is None
def get_root(self) -> "Project":
"""The root project of a vendor chain."""
if self.is_root():
return self
p = self._parent
while p is not None and not p.is_root():
p = p._parent
assert p is not None
return p
def get_vendors(self) -> list[str]:
"""Names of vendored dependencies under ``vendor/``."""
try:
return [
vendor
for vendor in os.listdir(self.path("vendor"))
if os.path.isdir(self.path("vendor", vendor))
# todo: how to ignore vendors?
]
except FileNotFoundError:
return []
def on_resolve_manifests(self) -> list[str]:
"""Code-provenance tools to capture for each run.
Returns importable ``Manifest`` modules; the default is the built-in
git manifest. Override in your project to add or replace tools (a
dependency lock, a container digest); each is an ordinary
:class:`~machinable.manifest.Manifest` subclass, resolved like any
interface.
"""
return ["machinable.manifest"]
def on_resolve_discovery(self) -> list:
"""Discovery schemes consulted when enumerating this project's modules.
Returns importable ``Discovery`` elements (a module string, a
``module:Class`` reference, or a class), resolved in order and composed
first-wins. The default is the built-in static AST discovery.
"""
return ["machinable.discovery", "machinable.discovery:RemoteDiscovery"]
def _resolve_discovery(self, element):
"""Resolve one ``on_resolve_discovery`` element to a ``Discovery``."""
from machinable.discovery import Discovery
if isinstance(element, Discovery):
return element
if isinstance(element, type):
return element()
if isinstance(element, str) and ":" in element:
from machinable.config import import_object_by_path
obj = import_object_by_path(element)
return obj() if isinstance(obj, type) else obj
mod = import_from_directory(element, self.config.directory)
if mod is None:
mod = importlib.import_module(element)
return find_subclass_in_module(mod, Discovery, default=Discovery)()
def modules(self) -> list:
"""Statically discovered interface modules."""
from machinable.discovery import Catalog
catalog = Catalog()
graph = None
for element in self.provider().on_resolve_discovery():
try:
scheme = self._resolve_discovery(element)
graph = scheme.discover(graph, project=self, catalog=catalog) or graph
except KeyboardInterrupt:
raise
except BaseException: # noqa: BLE001 - a scheme failing never breaks discovery
# Importing an arbitrary project module can raise more than Exceptions:
# pytest's importorskip raises Skipped (a BaseException), a module may
# call sys.exit(). One odd module must not 500 the whole index.
continue
return catalog.entries()
def module_schema(self, module: str):
"""The static config schema for one discovered ``module``."""
graph = None
error: Exception | None = None
for element in self.provider().on_resolve_discovery():
try:
scheme = self._resolve_discovery(element)
except Exception: # noqa: BLE001 - skip an unresolvable scheme
continue
if graph is None:
graph = scheme._graph(self.path())
try:
return scheme.module_schema(module, project=self, graph=graph)
except LookupError as ex:
error = ex
raise error or LookupError(f"No interface module '{module}'")
def interface(
self,
module: "str | Interface | type[Interface]",
version: VersionType = None,
base_class: Any = None,
**constructor_kwargs,
) -> Interface:
"""Resolve and instantiate an interface module from this project."""
module, interface_class = self.provider().on_resolve_interface(module)
module, version = extend(cast("str | Interface", module), version)
if not isinstance(module, str):
# interactive session interface
interface_class = module
module = "__session__" + interface_class.__name__
else:
# import from project
if interface_class is None:
interface_class = import_interface(self.path(), module, base_class)
interface = instantiate(
cast(str, module),
interface_class,
version,
**constructor_kwargs,
)
return interface
def exists(self) -> bool:
"""True when the project directory exists."""
return os.path.exists(self.config.directory)
def serialize(self) -> dict:
"""The project as a plain dict."""
return {
"directory": self.config.directory,
}
@classmethod
def unserialize(cls, serialized: dict) -> "Project":
"""Rebuild a project from ``serialize()`` output."""
return cls(serialized)
def get_host_info(self) -> dict:
"""Returned dictionary will be recorded as host information."""
return {
"network_name": platform.node(),
"hostname": socket.gethostname(),
"machine": platform.machine(),
"python_version": platform.python_version(),
"user": getpass.getuser(),
"argv": sys.argv,
"machinable_version": machinable.get_version(),
}
def fetch_remote(self, module: str) -> str | None:
"""Fetch a declared remote module into ``interface/remotes/``, no import.
Recursively fetches declared dependencies and returns the local file
path, or ``None`` when ``module`` is not a declared remote. Since
remote modules execute on import, fetching first (``machinable
fetch``) lets you inspect the code before anything runs.
"""
remotes = self.on_resolve_remotes()
if module not in remotes:
return None
remote = remotes[module]
if isinstance(remote, list | tuple):
remote, *dependencies = remote
for dependency in dependencies:
self.fetch_remote(dependency)
directory = self.path("interface/remotes")
remote_module = module.replace(".", "_")
filename = os.path.join(directory, remote_module + ".py")
# obtain from remote if not existing
if not os.path.exists(filename):
os.makedirs(directory, exist_ok=True)
if remote.startswith("url+"):
# download
with (
urllib.request.urlopen(remote[4:]) as response,
open(filename, "wb") as out_file,
):
data = response.read()
out_file.write(data)
elif remote.startswith("link+"):
# symlink, falling back to a copy where symlinks are not permitted
# (Windows without admin/Developer Mode) so the import still resolves
if os.path.islink(filename):
os.remove(filename)
try:
os.symlink(remote[5:], filename)
except OSError:
shutil.copy(remote[5:], filename)
elif remote.startswith("file+"):
# copy
shutil.copy(remote[5:], filename)
else:
raise ValueError(
f"Unknown remote type for {module}: {remote}"
f" (target file is {filename}"
)
return filename
def resolve_remotes(self, module: str) -> Interface | None:
"""Fetch and import a remote module declared by ``on_resolve_remotes``."""
remotes = self.on_resolve_remotes()
if module not in remotes:
return None
remote = remotes[module]
if isinstance(remote, list | tuple):
remote, *dependencies = remote
# dependencies must be imported (not just fetched) so the main
# module's bare-name imports resolve through sys.modules
for dependency in dependencies:
self.resolve_remotes(dependency)
filename = self.fetch_remote(module)
directory = self.path("interface/remotes")
remote_module = module.replace(".", "_")
try:
interface_class = find_subclass_in_module(
import_from_directory(remote_module, directory, or_fail=True),
Interface,
)
if interface_class is None:
raise ModuleNotFoundError(
"Could not find an interface in remote module"
)
return interface_class
except ModuleNotFoundError as _ex:
raise ValueError(
f"Invalid remote import: {remote} for {module} at {filename}"
) from _ex
def on_resolve_remotes(self) -> "dict[str, str | list[str]]":
"""Event triggered during remote resolution.
Return a dictionary of module names and their remote source (or a
``[source, *dependency_modules]`` list).
"""
return {}
def on_resolve_interface(
self, module: "str | Interface | type[Interface]"
) -> "tuple[str | Interface | type[Interface], type[Interface] | None]":
"""Override interface resolution.
Return altered module and/or resolved Interface class to be used instead.
"""
if isinstance(module, str):
try:
return self._cache["on_resolve_interface"][module]
except KeyError:
self._cache.setdefault("on_resolve_interface", {})
v = (
module,
cast("type[Interface] | None", self.resolve_remotes(module)),
)
self._cache["on_resolve_interface"][module] = v
return v
return module, None
def on_resolve_vendor(
self, name: str, source: str | None, target: str
) -> bool | None:
"""Event triggered when vendor is resolved.
# Arguments
name: The name of the vendor
source: The source configuration
target: The target directory (may or may not exists yet)
Return False to prevent the default automatic resolution
"""
def __repr__(self) -> str:
return f"Project({self.config.directory})"