"""Resolve the tracking context: which Hub, project, run and batch.
Every value is resolved through the same layers, highest first:
1. an argument (a CLI flag, or a keyword passed in Python),
2. the environment (``EMBEDL_HUB_PROJECT``, ``EMBEDL_HUB_RUN``, ...) —
how ``embedl-hub exec``, schedulers and CI hand context down,
3. the nearest workspace (``.ehub/``, created by ``embedl-hub init``),
4. for the project only, the global config.
There is deliberately no global run or batch: an ambient "current run"
shared by every console is how logs end up on the wrong run.
The CLI and :class:`~embedl_hub.tracking.Client` both resolve through
:func:`resolve_context`, so they cannot disagree.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from typing import Literal
from embedl_hub._internal import config as global_config_file
from embedl_hub._internal.config import load_global_config, resolve_base_url
from embedl_hub._internal.tracking.errors import TrackingContextError
from embedl_hub._internal.types import LocalPath
from embedl_hub._internal.workspace import Workspace, load_use_context
RUN_ENV_VAR_NAME = "EMBEDL_HUB_RUN"
PROJECT_ENV_VAR_NAME = "EMBEDL_HUB_PROJECT"
BATCH_ENV_VAR_NAME = "EMBEDL_HUB_BATCH"
# Set by `embedl-hub exec --safe` so that adopting the run it handed over
# tolerates an unreachable Hub, as creating that run did. It covers the
# handover only: whether the command's own logging tolerates failure is
# the command's decision, made with `safe_log`.
SAFE_ENV_VAR_NAME = "EMBEDL_HUB_SAFE"
# Set by `embedl-hub exec` when it skipped tracking, so the command it
# wraps never falls through to the workspace's run.
UNTRACKED_ENV_VAR_NAME = "EMBEDL_HUB_UNTRACKED"
SettingOrigin = Literal["argument", "env", "workspace", "global", "default"]
_HUB_ORIGINS: dict[str, SettingOrigin] = {
"env": "env",
"workspace": "workspace",
"config": "global",
"default": "default",
}
_NOUNS = {"project": "project name", "run": "run ID", "batch": "batch ID"}
[docs]
@dataclass(frozen=True)
class Setting:
"""One resolved value and where it came from.
A layer that is present but empty — a blank argument, or a variable
that is set but empty — has an `origin` but no `value`. That is
almost always a shell expansion that produced nothing, so it is
reported rather than skipped.
:ivar value: The resolved value, or ``None``.
:ivar origin: The layer that supplied it, or ``None`` when no layer
has one.
:ivar detail: The environment variable's name or the file's path,
when the origin has one.
"""
value: str | None
origin: SettingOrigin | None = None
detail: str | None = None
[docs]
@dataclass(frozen=True)
class Context:
"""The tracking context resolved from every layer.
:ivar hub: The Hub's address.
:ivar project: The project new runs start in.
:ivar run: The run to log to. ``EMBEDL_HUB_UNTRACKED`` shows up
here as an ``"env"`` setting with no value.
:ivar batch: The batch log commands queue into.
"""
hub: Setting
project: Setting
run: Setting
batch: Setting
[docs]
def resolve_context(
*,
project: str | None = None,
run: str | None = None,
batch: str | None = None,
) -> Context:
"""Resolve the tracking context, changing nothing.
Never raises for a missing value: an unset setting has no origin.
:param project: A project name that outranks every other layer.
:param run: A run ID that outranks every other layer.
:param batch: A batch ID that outranks every other layer.
:return: Each value together with where it came from.
:raises TrackingContextError: If the global config cannot be read.
"""
return _resolve_context(project=project, run=run, batch=batch)
def _resolve_context(
*,
project: str | None = None,
run: str | None = None,
batch: str | None = None,
start: LocalPath | None = None,
global_config: dict[str, str] | None = None,
) -> Context:
"""Resolve the context, with the seams the CLI and tests need.
:param start: Where the workspace lookup starts (default: the
current directory).
:param global_config: The global config already loaded, as the CLI
holds it; read from disk when ``None``.
"""
values, workspace = load_use_context(start)
if global_config is None:
global_config = load_global_config()
config = global_config
project_setting = _pick(project, PROJECT_ENV_VAR_NAME, values, workspace)
if project_setting is None:
stored = config.get("project") or config.get("project_name")
project_setting = (
Setting(stored, "global", str(global_config_file.CONFIG_FILE))
if stored
else Setting(None)
)
run_setting = _pick(run, RUN_ENV_VAR_NAME, None, None)
if run_setting is None and _is_set(UNTRACKED_ENV_VAR_NAME):
run_setting = Setting(None, "env", UNTRACKED_ENV_VAR_NAME)
if run_setting is None:
run_setting = _pick(None, None, values, workspace, key="run")
batch_setting = _pick(
batch, BATCH_ENV_VAR_NAME, values, workspace, key="batch"
)
server = resolve_base_url()
return Context(
hub=Setting(
server.value, _HUB_ORIGINS[server.source], server.origin or None
),
project=project_setting,
run=run_setting or Setting(None),
batch=batch_setting or Setting(None),
)
def require_value(setting: Setting, kind: str) -> str | None:
"""Return a setting's value, rejecting a layer that is present but empty.
:param setting: The setting to read.
:param kind: ``"project"``, ``"run"`` or ``"batch"``, for the message.
:return: The value, or ``None`` when no layer has one.
:raises TrackingContextError: If an argument or variable is empty.
"""
if setting.value is not None or setting.origin is None:
return setting.value
if setting.detail == UNTRACKED_ENV_VAR_NAME:
return None
noun = _NOUNS[kind]
if setting.origin == "env":
raise TrackingContextError(
f"{setting.detail} is set but empty. This usually means a "
"shell capture failed before it was exported; unset the "
"variable or give it a value."
)
raise TrackingContextError(
f"The {noun} is empty. This usually means a shell variable "
f"expanded to nothing; pass a {noun}."
)
def is_untracked(setting: Setting) -> bool:
"""Whether `embedl-hub exec` marked this process as untracked."""
return setting.value is None and setting.detail == UNTRACKED_ENV_VAR_NAME
def _pick(
argument: str | None,
env_var: str | None,
values: dict[str, str] | None,
workspace: Workspace | None,
key: str = "project",
) -> Setting | None:
"""Take the first layer that has `key`: argument, env, workspace."""
if argument is not None:
return Setting(argument if argument.strip() else None, "argument")
if env_var is not None and (raw := os.getenv(env_var)) is not None:
return Setting(raw if raw.strip() else None, "env", env_var)
if workspace is not None and values and (stored := values.get(key)):
return Setting(stored, "workspace", str(workspace.source_of(key)))
return None
def _is_set(env_var: str) -> bool:
return bool((os.getenv(env_var) or "").strip())
