fractal.util package#

Submodules#

fractal.util.duration module#

Functions for parsing and formatting durations.

fractal.util.duration.parse_duration_seconds(value)[source]#

Parse a duration string (e.g. 30s, 10m, 1.5h, 2d) into seconds.

Parameters:: value (str) – Duration string of the form <number><s|m|h|d>.
Return type:: float | None
Returns:: Seconds, or None when value is not <number><s|m|h|d> – callers decide whether to raise or fall back.

Examples

>>> parse_duration_seconds('30s')
30.0
>>> parse_duration_seconds('1.5h')
5400.0
>>> parse_duration_seconds('soon') is None
True

fractal.util.duration.format_age(seconds)[source]#

Render elapsed seconds as a compact age (43s / 18m / 1h).

Parameters:: seconds (float) – Elapsed seconds.
Return type:: str
Returns:: The age truncated to its largest unit (seconds, minutes, or hours).

Examples

>>> format_age(43.0)
'43s'
>>> format_age(1_080.0)
'18m'
>>> format_age(90_000.0)
'25h'

fractal.util.filesystem module#

Functions for saving and loading on the file system.

fractal.util.filesystem.write_atomic(path, data)[source]#

Write data to path atomically (unique temp + os.replace).

A plain write_text truncates before writing, so a concurrent reader can observe an empty or partial file and a crash mid-write leaves a torn one. Staging to a dot-prefixed temp file in the target’s directory and swapping it into place makes every read all-or-nothing. Text writes land utf-8 with LF endings verbatim. A symlinked destination resolves first, so the swap updates the link’s target and the link itself survives. mkstemp creates the temp 0600 and os.replace carries that mode onto the target, so an existing target’s mode is preserved explicitly and a fresh file honors the umask.

Parameters:

path (Path) – Destination file.
data (str | bytes) – Payload – str lands utf-8 with LF endings, bytes verbatim.

Return type:

None

fractal.util.git module#

Functions for local git operations via subprocess.

fractal.util.git.run(cmd, *, cwd=None, check=True)[source]#

Run a git command and return stripped stdout.

Overloads:

cmd (list[str]), cwd (Optional[pathlib.Path]), check (Literal[True]) → str
cmd (list[str]), cwd (Optional[pathlib.Path]), check (Literal[False]) → Optional[str]

Parameters:

cmd (list[str]) – Git subcommand and arguments (without git prefix).
cwd (Path | None) – Working directory for the command.
check (bool) – Raise RuntimeError on non-zero exit.

Returns:

Stripped stdout string, or None on non-zero exit when check is False.

fractal.util.git.run_bytes(cmd, *, cwd=None)[source]#

Run a git command and return raw stdout bytes (None on failure).

The binary-safe, unstripped, non-raising counterpart to run(), for reading a file blob (git show <ref>:<path>) or a NUL-separated listing where text decoding and whitespace stripping would corrupt content – a non-zero exit (e.g. the path absent at that revision) degrades to None.

Parameters:

cmd (list[str]) – Git subcommand and arguments (without git prefix).
cwd (Path | None) – Working directory for the command.

Return type:

bytes | None

Returns:

Raw stdout bytes, or None on non-zero exit.

fractal.util.git.toplevel(path, *, check=True)[source]#

Return the worktree top-level directory for a path.

Parameters:

path (Path) – A directory inside the worktree.
check (bool) – Raise RuntimeError on non-zero exit.

Return type:

Path | None

Returns:

The worktree top-level, or None on non-zero exit when check is False.

fractal.util.git.common_dir(path)[source]#

Return the main git repo root (resolves through worktrees).

Parameters:: path (Path) – A directory inside any of the repo’s worktrees.
Return type:: Path
Returns:: The main repository root.

fractal.util.git.branch(path, *, check=True)[source]#

Return the checked-out branch name for a worktree.

Parameters:

path (Path) – A directory inside the worktree.
check (bool) – Raise RuntimeError on non-zero exit.

Return type:

str | None

Returns:

The branch name, or None on non-zero exit when check is False.

fractal.util.git.worktree_map(repo_dir)[source]#

Map each branch to its on-disk worktree path from one git worktree list.

The batched form of find_worktree() – callers resolving many branches (e.g. a listing decorating a whole subtree) build the map once instead of spawning a git worktree list per branch. A worktree whose directory no longer exists (rm -rf’d out of band, still listed by git as prunable) is dropped, so both the listing and the resolver agree a hand-removed node is gone.

Parameters:: repo_dir (Path) – Main repo root.
Return type:: dict[str, Path]
Returns:: Mapping of branch name to worktree path.

fractal.util.git.find_worktree(repo_dir, branch)[source]#

Find the on-disk worktree path for a branch.

A thin lookup over worktree_map(), so it shares the same on-disk probe – a worktree rm -rf’d out of band still lists (as prunable) in git’s porcelain, but its directory is gone, so it resolves as absent here rather than handing back a dead path.

Parameters:

repo_dir (Path) – Main repo root.
branch (str) – Branch name.

Return type:

Path | None

Returns:

Worktree path, or None if not found.

fractal.util.git.prunable(repo_dir)[source]#

Return whether git lists any prunable (dead-on-disk) worktree.

A worktree rm -rf’d out of band lingers in git worktree list as prunable until git worktree prune clears its metadata (and the lingering entry keeps its branch ref checked out, resisting deletion) – callers use this to point at that one-shot cleanup.

Parameters:: repo_dir (Path) – Main repo root.
Return type:: bool
Returns:: Whether at least one listed worktree is prunable.

fractal.util.system module#

Functions for locating the running installation.

fractal.util.system.bin_dir()[source]#

Return the invoking interpreter’s script directory.

Return type:: str

fractal.util.system.console_script(name)[source]#

Resolve a console script robustly.

Prefers the script beside the running interpreter – a dependency’s console script installs into the same environment – and falls back to the ambient PATH. Ambient-only resolution breaks under pyenv shims: an orphaned shim resolves but exits 127 with a confusing “command not found” message.

Parameters:: name (str) – Console script name.
Return type:: str
Returns:: Path of the executable.
Raises:: RuntimeError – If no executable exists.

fractal.util.system.prepend_bin_path(env=None)[source]#

Return an environment with bin_dir() prepended to PATH.

Scripts that shell back into the installation’s own CLIs must resolve them from the invoking installation, not ambient PATH, so a fronted foreign install (e.g. the root venv’s) cannot answer in this one’s place.

Parameters:: env (dict[str, str] | None) – Base environment (defaults to os.environ).
Return type:: dict[str, str]
Returns:: A fresh environment mapping with the prepended PATH.

fractal.util.time module#

Functions for UTC timestamps and elapsed durations.

fractal.util.time.utc_now()[source]#

Return current UTC time as ISO 8601 (millisecond precision).

Matches the strftime('%Y-%m-%dT%H:%M:%fZ') format used by the SQL created_at defaults (core/schema.sql), so the Python-sourced started_at/ended_at stamps share the same precision and sort against them.

Return type:: str

fractal.util.time.elapsed(started_at)[source]#

Compute seconds elapsed since a start timestamp.

Parameters:: started_at (str) – ISO 8601 timestamp with fractional seconds.
Return type:: float
Returns:: Elapsed seconds as a float.

Examples

>>> 0.0 <= elapsed(utc_now()) < 1.0
True

fractal.util.title module#

Functions for deriving display titles.

fractal.util.title.name_to_title(name)[source]#

Turn a node-name slug into a display title (foo_bar -> Foo Bar).

Parameters:: name (str) – A node name – a leaf slug of letters, digits, and underscores.
Return type:: str
Returns:: The de-slugged, title-cased name.

Examples

>>> name_to_title('foo_bar')
'Foo Bar'
>>> name_to_title('foo_')
'Foo'

fractal.util.tmux module#

Functions for tmux session probing.

fractal.util.tmux.probe(*, socket=None)[source]#

Return the set of live tmux session names, or None when unknown.

The batched form of a per-session existence probe – a caller checking many sessions (e.g. reconciling a whole subtree) probes once instead of per name. Empty only when tmux definitively answered “no sessions”: a zero exit, the no server running refusal, or a connect against a socket path that does not exist (no server ever started on this socket). Any other failure – the binary absent (OSError, raised before any result) or list-sessions erroring for another reason (socket permissions, an unexpected refusal) – is inconclusive and returns None, so a caller about to act destructively on “no sessions” can refuse to act on ignorance.

Every answer is evidence about one server only: the ambient socket’s “no sessions” says nothing about sessions alive on another socket, so a caller judging a specific session passes the socket it lives on.

Parameters:: socket (str | None) – Server socket path to probe (tmux -S); None probes the ambient socket.
Return type:: frozenset[str] | None
Returns:: The live tmux session names, or None when tmux gave no answer.

fractal.util.tmux.sessions()[source]#

Return the set of live tmux session names (one list-sessions).

probe() with the inconclusive answer folded into the empty set – for display-only callers where “unknown” and “none visible” render the same and a failed probe must never crash the read.

Return type:: frozenset[str]
Returns:: The live tmux session names (empty when the probe failed).

Module contents#

Utilities for fractal.