work on runtime flows, exit code handling, chaining, and implement basic assertion functions

pyproject.toml

@@ -8,7 +8,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "dotdir"
-version = "0.1.0"
+version = "0.2.0"
 description = "Dotdir is the daughter of my old 'dot file' repository and setup script manager."
 readme = "README.md"
 authors = [{ name = "Marcel Simader", email = "marcel0simader@gmail.com" }]
@@ -19,6 +19,7 @@ license = {text = "Proprietary"}
 [tool.setuptools]
 packages = ["dotdir"]
 package-dir = { "" = "src" }
+package-data = { "dotdir" = ['py.typed'] }
 
 [project.scripts]
 dotdir = "dotdir.main:main"

setup.cfg

@@ -4,5 +4,5 @@
 
 [pycodestyle]
 exclude = .git,dist,test_distro
-ignore = E302,E305,E701,E261
+ignore = E302,E305,E701,E261,F403,F405,F406
 

src/dotdir/__init__.py

@@ -7,7 +7,9 @@ from .io import (
     DotdirTreeFormatter as DotdirTreeFormatter,
 )
 from .runtime import (
+    ChainContextManager as ChainContextManager,
     Command as Command,
+    Config as Config,
     Context as Context,
     IntProvider as IntProvider,
     PathContextManager as PathContextManager,

src/dotdir/__init__.pyi

@@ -2,21 +2,32 @@
 #  Date: 15.11.2023
 #  (c) Marcel Simader 2023
 
-from typing import Sequence, TypeGuard, overload, Iterator
-
-from pathlib import Path
-from os import _Environ as EnvironType
 import subprocess
-
+from os import _Environ as EnvironType
+from pathlib import Path
+from typing import Iterator, Sequence, TypeGuard, overload
+
+from .io import DotdirTreeFormatter, EscSeq, BRIGHT_BLACK
+from .protocols import (
+    SupportsContains,
+    SupportsEQ,
+    SupportsGE,
+    SupportsGT,
+    SupportsLE,
+    SupportsLT,
+    SupportsNE,
+    SupportsTruthy,
+)
 from .runtime import (
+    Command as Command,
     Context as Context,
     IntProvider as IntProvider,
-    Command as Command,
-    SomePath as SomePath,
     PathContextManager as PathContextManager,
+    SomePath as SomePath,
     SortingOption as SortingOption,
+    Config as Config,
+    ChainContextManager as ChainContextManager,
 )
-from .io import DotdirTreeFormatter
 
 #  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 #  ~~~~~~~~~~~~~~~~~~~~ Values ~~~~~~~~~~~~~~~~~~~~
@@ -28,12 +39,18 @@ pwd: Path
 current_path: Path
 code: int
 environ: EnvironType
+config: Config
 
 #  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 #  ~~~~~~~~~~~~~~~~~~~~ Builtin Overloads ~~~~~~~~~~~~~~~~~~~~
 #  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-def print(*values: object, sep: str = " ", end: str = "\n") -> None:
+def print(
+    *values: object,
+    style: EscSeq = BRIGHT_BLACK,
+    sep: str = " ",
+    end: str = "\n"
+) -> None:
     ...
 
 #  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -118,3 +135,60 @@ def git_pull(dir: SomePath, *args: Command) -> bool:
 
 def git_ensure(uri: str, dest: SomePath, *args: Command) -> bool:
     ...
+
+#  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+#  ~~~~~~~~~~~~~~~~~~~~ Flow ~~~~~~~~~~~~~~~~~~~~
+#  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+def chain(description: str = "", /) -> ChainContextManager:
+    ...
+
+#  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+#  ~~~~~~~~~~~~~~~~~~~~ Assertions ~~~~~~~~~~~~~~~~~~~~
+#  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+#  ~~~~~~~~~~~~~~~~~~~~ Basic Assertions ~~~~~~~~~~~~~~~~~~~~
+
+def assert_true(o: SupportsTruthy, /, msg: str = "") -> None:
+    ...
+
+def assert_false(o: SupportsTruthy, /, msg: str = "") -> None:
+    ...
+
+def assert_is(o1: object, o2: object, /, msg: str = "") -> None:
+    ...
+
+def assert_is_not(o1: object, o2: object, /, msg: str = "") -> None:
+    ...
+
+def assert_eq(o1: SupportsEQ, o2: SupportsEQ, /, msg: str = "") -> None:
+    ...
+
+def assert_ne(o1: SupportsNE, o2: SupportsNE, /, msg: str = "") -> None:
+    ...
+
+def assert_lt(o1: SupportsLT, o2: SupportsLT, /, msg: str = "") -> None:
+    ...
+
+def assert_gt(o1: SupportsGT, o2: SupportsGT, /, msg: str = "") -> None:
+    ...
+
+def assert_le(o1: SupportsLE, o2: SupportsLE, /, msg: str = "") -> None:
+    ...
+
+def assert_ge(o1: SupportsGE, o2: SupportsGE, /, msg: str = "") -> None:
+    ...
+
+def assert_in(o1: object, o2: SupportsContains, /, msg: str = "") -> None:
+    ...
+
+def assert_not_in(o1: object, o2: SupportsContains, /, msg: str = "") -> None:
+    ...
+
+#  ~~~~~~~~~~~~~~~~~~~~ Path Assertions ~~~~~~~~~~~~~~~~~~~~
+
+def assert_file(path: SomePath, /, *, deref: bool = True) -> None:
+    ...
+
+def assert_dir(path: SomePath, /, *, deref: bool = True) -> None:
+    ...

src/dotdir/codes.py

+#  Author: Marcel Simader (marcel0simader@gmail.com)
+#  Date: 19.11.2023
+#  (c) Marcel Simader 2023
+
+from enum import IntEnum
+
+#  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+#  ~~~~~~~~~~~~~~~~~~~~ Exit Codes ~~~~~~~~~~~~~~~~~~~~
+#  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+class ExitCode(IntEnum):
+
+    ASSERTION_FAILED = 70

src/dotdir/dotdir.py

@@ -9,11 +9,28 @@ from collections import Counter
 from pathlib import Path
 from typing import Any, TypeVar
 
-from .executors import (CreateDirExecutor, CreateFileExecutor, Executor,
-                        GlobLinkExecutor, ProcessExecutor, PythonExecutor)
-from .io import (BRIGHT_BLACK, BRIGHT_GREEN, BRIGHT_RED, GREEN, LOG, NORMAL,
-                 RED, UNDERLINE, YELLOW, DotdirFormatter,
-                 DotdirFormatterManager, DotdirTreeFormatter)
+from .executors import (
+    CreateDirExecutor,
+    CreateFileExecutor,
+    Executor,
+    GlobLinkExecutor,
+    ProcessExecutor,
+    PythonExecutor,
+)
+from .io import (
+    BRIGHT_BLACK,
+    BRIGHT_GREEN,
+    BRIGHT_RED,
+    GREEN,
+    LOG,
+    NORMAL,
+    RED,
+    UNDERLINE,
+    YELLOW,
+    DotdirFormatter,
+    DotdirFormatterManager,
+    DotdirTreeFormatter,
+)
 from .seconf import parse_seconf_file
 
 #  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

src/dotdir/executors/py.py

@@ -37,7 +37,7 @@ class PythonExecutor(Executor[subprocess.CompletedProcess]):
                 file_contents = file.read()
 
             sub_globals: dict[str, Any] = dict()
-            ctx = runtime.Context(sub_globals, self._tfmt, Path.cwd(), 0)
+            ctx = runtime.Context(sub_globals, self._tfmt, Path.cwd())
             sub_globals.update({
                 "ctx": ctx,
                 **runtime.bind_context_globals(ctx),
@@ -45,6 +45,8 @@ class PythonExecutor(Executor[subprocess.CompletedProcess]):
             })
             try:
                 exec(file_contents, sub_globals)
+            except runtime.asserts.DotdirAssertion as ass:
+                ass.trigger()
             except ImportError as err:
                 if (err.path is None) or (err.path.index("dotdir") == -1):
                     raise err
@@ -55,7 +57,10 @@ class PythonExecutor(Executor[subprocess.CompletedProcess]):
                     " Note, that it is still possible to import from"
                     " 'dotdir.io', etc.",
                 ) from err
-            return subprocess.CompletedProcess((), sub_globals["ctx"].code)
+            return subprocess.CompletedProcess(
+                (),
+                sub_globals["ctx"].pop_final_code(),
+            )
         except Exception as err:
             LOG.exception(f"Error while running Python process: {err!s}")
             raise err

src/dotdir/protocols.py

+#  Author: Marcel Simader (marcel0simader@gmail.com)
+#  Date: 19.11.2023
+#  (c) Marcel Simader 2023
+
+from typing import Protocol, SupportsInt, TypeAlias, runtime_checkable
+
+#  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+#  ~~~~~~~~~~~~~~~~~~~~ Protocols ~~~~~~~~~~~~~~~~~~~~
+#  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+@runtime_checkable
+class SupportsBool(Protocol):
+
+    def __bool__(self) -> bool:
+        ...
+
+SupportsTruthy: TypeAlias = SupportsInt | SupportsBool
+
+@runtime_checkable
+class SupportsEQ(Protocol):
+
+    def __eq__(self, o: object) -> bool:
+        ...
+
+@runtime_checkable
+class SupportsNE(Protocol):
+
+    def __ne__(self, o: object) -> bool:
+        ...
+
+@runtime_checkable
+class SupportsLT(Protocol):
+
+    def __lt__(self, o: object) -> bool:
+        ...
+
+@runtime_checkable
+class SupportsGT(Protocol):
+
+    def __gt__(self, o: object) -> bool:
+        ...
+
+@runtime_checkable
+class SupportsLE(Protocol):
+
+    def __le__(self, o: object) -> bool:
+        ...
+
+@runtime_checkable
+class SupportsGE(Protocol):
+
+    def __ge__(self, o: object) -> bool:
+        ...
+
+@runtime_checkable
+class SupportsContains(Protocol):
+
+    def __contains__(self, o: object) -> bool:
+        ...

src/dotdir/runtime/__init__.py

@@ -13,11 +13,14 @@ import dotdir.runtime.builtin_overloads
 import dotdir.runtime.perform
 import dotdir.runtime.git
 import dotdir.runtime.paths
+import dotdir.runtime.flow
+import dotdir.runtime.asserts
 
 # NOTE: These are the exports we actually need to access by name.
 
 from .base import (
     Command,
+    Config,
     Context,
     IntProvider,
     SomePath,
@@ -28,3 +31,6 @@ from .paths import (
     PathContextManager,
     SortingOption,
 )
+from .flow import (
+    ChainContextManager,
+)

src/dotdir/runtime/asserts.py

+#  Author: Marcel Simader (marcel0simader@gmail.com)
+#  Date: 19.11.2023
+#  (c) Marcel Simader 2023
+
+from enum import Enum
+from typing import Generic, TypeVar
+
+from ..codes import ExitCode
+from ..io import BRIGHT_RED
+from ..protocols import (
+    SupportsBool,
+    SupportsContains,
+    SupportsEQ,
+    SupportsGE,
+    SupportsGT,
+    SupportsLE,
+    SupportsLT,
+    SupportsNE,
+    SupportsTruthy,
+)
+from .base import (
+    Context,
+    SomePath,
+    _mark_runtime_partial,
+    _RuntimeCallable,
+)
+from .paths import _rf_is_dir, _rf_is_file
+
+#  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+#  ~~~~~~~~~~~~~~~~~~~~ Basic Assertions ~~~~~~~~~~~~~~~~~~~~
+#  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+_A = TypeVar("_A")
+_B = TypeVar("_B")
+
+class Comparison(Enum):
+    IS = (lambda a, b: a is b, "is")
+    IS_NOT = (lambda a, b: a is not b, "is not")
+    EQ = (lambda a, b: a == b, "==")
+    NE = (lambda a, b: a != b, "!=")
+    GT = (lambda a, b: a > b, ">")
+    LT = (lambda a, b: a < b, "<")
+    GE = (lambda a, b: a >= b, ">=")
+    LE = (lambda a, b: a <= b, "<=")
+    IN = (lambda a, b: a in b, "in")
+    NOT_IN = (lambda a, b: a not in b, "not in")
+
+    def test(self, a: object, b: object, /) -> bool:
+        return self.value[0](a, b)
+
+    @property
+    def symbol(self) -> str:
+        return self.value[1]
+
+class DotdirAssertion(RuntimeError, Generic[_A, _B]):
+
+    ctx: Context
+    a: _A
+    b: _B
+    expansion: str
+
+    def __init__(
+        self, ctx: Context, a: _A, comp: Comparison, b: _B, message: str = ""
+    ):
+        expansion = f"{a!r} {comp.symbol} {b!r} FAILED" \
+            f"{': ' if len(message) > 0 else ''}{message}"
+        super(DotdirAssertion, self).__init__(expansion)
+        self.ctx = ctx
+        self.a = a
+        self.b = b
+        self.expansion = expansion
+
+    def trigger(self) -> None:
+        self.ctx.tfmt.expand(BRIGHT_RED, self.expansion)
+        self.ctx.push_code(ExitCode.ASSERTION_FAILED)
+        self.ctx.merge_codes()
+
+@_mark_runtime_partial
+def _rp_assert_true(ctx: Context, o: SupportsTruthy, /, msg: str = "") -> None:
+    if not o: raise DotdirAssertion(ctx, bool(o), Comparison.EQ, True, msg)
+
+@_mark_runtime_partial
+def _rp_assert_false(
+    ctx: Context, o: SupportsTruthy, /, msg: str = ""
+) -> None:
+    if o: raise DotdirAssertion(ctx, bool(o), Comparison.EQ, False, msg)
+
+@_mark_runtime_partial
+def _rp_assert_is(
+    ctx: Context, o1: object, o2: object, /, msg: str = ""
+) -> None:
+    if o1 is not o2: raise DotdirAssertion(ctx, o1, Comparison.IS, o2, msg)
+
+@_mark_runtime_partial
+def _rp_assert_is_not(
+    ctx: Context, o1: object, o2: object, /, msg: str = ""
+) -> None:
+    if o1 is o2: raise DotdirAssertion(ctx, o1, Comparison.IS_NOT, o2, msg)
+
+@_mark_runtime_partial
+def _rp_assert_eq(
+    ctx: Context, o1: SupportsEQ, o2: SupportsEQ, /, msg: str = ""
+) -> None:
+    if o1 != o2: raise DotdirAssertion(ctx, o1, Comparison.EQ, o2, msg)
+
+@_mark_runtime_partial
+def _rp_assert_ne(
+    ctx: Context, o1: SupportsNE, o2: SupportsNE, /, msg: str = ""
+) -> None:
+    if o1 == o2: raise DotdirAssertion(ctx, o1, Comparison.NE, o2, msg)
+
+@_mark_runtime_partial
+def _rp_assert_lt(
+    ctx: Context, o1: SupportsLT, o2: SupportsLT, /, msg: str = ""
+) -> None:
+    if o1 < o2: raise DotdirAssertion(ctx, o1, Comparison.LT, o2, msg)
+
+@_mark_runtime_partial
+def _rp_assert_gt(
+    ctx: Context, o1: SupportsGT, o2: SupportsGT, /, msg: str = ""
+) -> None:
+    if o1 > o2: raise DotdirAssertion(ctx, o1, Comparison.GT, o2, msg)
+
+@_mark_runtime_partial
+def _rp_assert_le(
+    ctx: Context, o1: SupportsLE, o2: SupportsLE, /, msg: str = ""
+) -> None:
+    if o1 <= o2: raise DotdirAssertion(ctx, o1, Comparison.LE, o2, msg)
+
+@_mark_runtime_partial
+def _rp_assert_ge(
+    ctx: Context, o1: SupportsGE, o2: SupportsGE, /, msg: str = ""
+) -> None:
+    if o1 >= o2: raise DotdirAssertion(ctx, o1, Comparison.GE, o2, msg)
+
+@_mark_runtime_partial
+def _rp_assert_in(
+    ctx: Context, o1: object, o2: SupportsContains, /, msg: str = ""
+) -> None:
+    if o1 not in o2: raise DotdirAssertion(ctx, o1, Comparison.IN, o2, msg)
+
+@_mark_runtime_partial
+def _rp_assert_not_in(
+    ctx: Context, o1: object, o2: SupportsContains, /, msg: str = ""
+) -> None:
+    if o1 in o2: raise DotdirAssertion(ctx, o1, Comparison.NOT_IN, o2, msg)
+
+#  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+#  ~~~~~~~~~~~~~~~~~~~~ Path Assertions ~~~~~~~~~~~~~~~~~~~~
+#  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+@_mark_runtime_partial
+def _rp_assert_file(
+    ctx: Context, path: SomePath, /, *, deref: bool = True
+) -> None:
+    if not _rf_is_file(path):
+        raise DotdirAssertion(
+            ctx, path, Comparison.IS, "File", "File not found"
+        )
+
+@_mark_runtime_partial
+def _rp_assert_dir(
+    ctx: Context, path: SomePath, /, *, deref: bool = True
+) -> None:
+    if not _rf_is_dir(path):
+        raise DotdirAssertion(
+            ctx, path, Comparison.IS, "Directory", "Directory not found"
+        )

src/dotdir/runtime/base.py

@@ -5,13 +5,15 @@
 from __future__ import annotations
 
 import os
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+from enum import Flag, auto
 from functools import partial
 from os import PathLike, _Environ as EnvironType
 from pathlib import Path
 from typing import (
     Any,
     Callable,
+    ClassVar,
     Concatenate,
     Mapping,
     ParamSpec,
@@ -33,22 +35,67 @@ SomePath: TypeAlias = str | PathLike | Path
 #  ~~~~~~~~~~~~~~~~~~~~ Runtime Context ~~~~~~~~~~~~~~~~~~~~
 #  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
+class StopDotdirChain(Exception):
+    pass
+
+@final
+class Config(Flag):
+    _ = 0
+
 @final
 @dataclass(init=True, repr=True, slots=True)
 class Context:
-    _globals: dict[str, Any]
-    _tfmt: DotdirTreeFormatter
-    _pwd: Path
-    _code: int
 
-    @property
-    def tfmt(self) -> DotdirTreeFormatter:
-        return self._tfmt
+    BASE_CODE_LEVEL: ClassVar[int] = 1
+
+    # NOTE: Counter-intuitively, the attributes *without* '_' are our private
+    #       context variables. This is because we automatically register
+    #       attributes defined as properties as context globals.
+    #
+    #       A user would need to type 'ctx.tfmt' to access 'tfmt', for
+    #       instance, instead of simply typing 'tfmt'.
 
-    @tfmt.setter
-    def tfmt(self, tfmt: DotdirTreeFormatter) -> None:
-        self._tfmt = tfmt
-        self._globals["tfmt"] = tfmt
+    globals_: dict[str, Any]
+    tfmt: DotdirTreeFormatter
+    code_stack: list[int] = field(
+        default_factory=lambda: Context.BASE_CODE_LEVEL * [0], init=False
+    )
+
+    # NOTE: These are the property-based context globals.
+
+    _pwd: Path
+    _config: Config = field(default=Config(0), init=False)
+
+    def push_code(self, code: int) -> None:
+        self.code_stack.append(code)
+
+    def pop_intermediate_code(self) -> int:
+        if len(self.code_stack) < Context.BASE_CODE_LEVEL:
+            raise IndexError(
+                "Context code stack empty! This should not be possible."
+            )
+        return self.code_stack.pop()
+
+    def pop_final_code(self) -> int:
+        if len(self.code_stack) > Context.BASE_CODE_LEVEL:
+            raise IndexError(
+                "Context code stack contains more than one item for final"
+                " pop! This should not be possible."
+            )
+        return self.pop_intermediate_code()
+
+    def merge_codes(self) -> None:
+        if len(self.code_stack) < 2:
+            raise IndexError(
+                "Context code stack must contain at least"
+                " 2 elements for a merge!"
+            )
+        a, b = self.pop_intermediate_code(), self.pop_intermediate_code()
+        c = a if (not b) else b
+        self.push_code(c)
+        print(len(self.code_stack))
+        if c and len(self.code_stack) > Context.BASE_CODE_LEVEL:
+            raise StopDotdirChain
 
     @property
     def pwd(self) -> Path:
@@ -57,26 +104,26 @@ class Context:
     @pwd.setter
     def pwd(self, pwd: Path) -> None:
         self._pwd = pwd
-        self._globals["pwd"] = pwd
-        self._globals["current_path"] = pwd
+        self.globals_["pwd"] = pwd
+        self.globals_["current_path"] = pwd
 
     @property
     def current_path(self) -> Path:
         return self._pwd
 
-    @property
-    def code(self) -> int:
-        return self._code
-
-    @code.setter
-    def code(self, code: int) -> None:
-        self._code = code
-        self._globals["code"] = code
-
     @property
     def environ(self) -> EnvironType:
         return os.environ
 
+    @property
+    def config(self) -> Config:
+        return self._config
+
+    @config.setter
+    def config(self, config: Config) -> None:
+        self._config = config
+        self.globals_["config"] = config
+
 _RUNTIME_CTX_GLOBALS: list[str] = []
 
 for member_name in dir(Context):

src/dotdir/runtime/builtin_overloads.py

@@ -5,7 +5,7 @@
 import sys
 from typing import Callable, Final
 
-from ..io import BRIGHT_BLACK
+from ..io import BRIGHT_BLACK, EscSeq
 from .base import Context, _mark_runtime_partial
 
 #  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -23,7 +23,11 @@ _builtin_print: Final[Callable] = print
 
 @_mark_runtime_partial
 def _rp_print(
-    ctx: Context, *values: object, sep: str = " ", end: str = "\n"
+    ctx: Context,
+    *values: object,
+    style: EscSeq = BRIGHT_BLACK,
+    sep: str = " ",
+    end: str = "\n"
 ) -> None:
     stringified = sep.join(str(o) for o in values)
     _builtin_print(stringified, sep=sep, end=end, file=sys.stdout)

src/dotdir/runtime/flow.py

+#  Author: Marcel Simader (marcel0simader@gmail.com)
+#  Date: 19.11.2023
+#  (c) Marcel Simader 2023
+
+from typing import Self
+
+from ..io import BRIGHT_GREEN, YELLOW, DotdirTreeFormatter
+from .base import Context, StopDotdirChain, _mark_runtime_partial
+
+#  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+#  ~~~~~~~~~~~~~~~~~~~~ Chain ~~~~~~~~~~~~~~~~~~~~
+#  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+class ChainContextManager:
+
+    _ctx: Context
+    _description: str
+    _old_tfmt: DotdirTreeFormatter
+
+    def __init__(self, ctx: Context, description: str = ""):
+        self._ctx = ctx
+        self._description = description
+        self._old_tfmt = self._ctx.tfmt
+
+    @property
+    def description(self) -> str:
+        return self._description
+
+    @property
+    def old_tfmt(self) -> DotdirTreeFormatter:
+        return self._old_tfmt
+
+    def __enter__(self) -> Self:
+        self._old_tfmt = self._ctx.tfmt
+        self._ctx.tfmt = self._ctx.tfmt.tree_formatter()
+        self._ctx.push_code(0)
+        if len(self._description) > 0:
+            self._ctx.tfmt.heading(
+                BRIGHT_GREEN, f"Chain {self._description!r}"
+            )
+        else:
+            self._ctx.tfmt.heading(BRIGHT_GREEN, "Chain")
+        return self
+
+    def __exit__(self, exc_type, exc_value, tb) -> bool:
+        self._ctx.merge_codes()
+        handled = False
+        if exc_type is StopDotdirChain:
+            # Handle exception
+            self._ctx.tfmt.expand(YELLOW, "Skipped rest of chain...")
+            handled |= True
+        self._ctx.tfmt.end()
+        self._ctx.tfmt = self._old_tfmt
+        return handled
+
+@_mark_runtime_partial
+def _rp_chain(ctx: Context, description: str = "", /) -> ChainContextManager:
+    return ChainContextManager(ctx, description)

src/dotdir/runtime/paths.py

@@ -64,21 +64,15 @@ class PathContextManager:
         _rp_cd(self._ctx, self._old_path)
         return None
 
-@runtime_checkable
-class SupportsLt(Protocol):
-
-    def __lt__(self, o: object) -> bool:
-        ...
-
 class SortingOption(Enum):
 
-    NAME = lambda p: p.name
-    SIZE = lambda p: p.stat().st_size
-    CREATION_TIME = lambda p: p.stat().st_ctime_ns
-    MODIFICATION_TIME = lambda p: p.stat().st_mtime_ns
-    ACCESS_TIME = lambda p: p.stat().st_atime_ns
-    DIRS_FIRST = lambda p: -p.is_dir()
-    FILES_FIRST = lambda p: p.is_dir()
+    NAME = (lambda p: p.name,)
+    SIZE = (lambda p: p.stat().st_size,)
+    CREATION_TIME = (lambda p: p.stat().st_ctime_ns,)
+    MODIFICATION_TIME = (lambda p: p.stat().st_mtime_ns,)
+    ACCESS_TIME = (lambda p: p.stat().st_atime_ns,)
+    DIRS_FIRST = (lambda p: -p.is_dir(),)
+    FILES_FIRST = (lambda p: p.is_dir(),)
 
 @_mark_runtime_partial
 def _rp_cd(ctx: Context, path: SomePath) -> PathContextManager:
@@ -99,12 +93,15 @@ def _rp_glob(
     pattern: str,
     *,
     recursively: bool = False,
-    sort_by: Callable[[Any], SupportsLt] | None = None,
+    sort_by: SortingOption | None = None,
     ascending: bool = True,
 ) -> Iterator[Path]:
     iterator = (ctx.pwd.rglob if recursively else ctx.pwd.glob)(pattern)
     if sort_by is None: yield from iterator
-    else: yield from sorted(iterator, key=sort_by, reverse=(not ascending))
+    else:
+        yield from sorted(
+            iterator, key=sort_by.value[0], reverse=(not ascending)
+        )
 
 #  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 #  ~~~~~~~~~~~~~~~~~~~~ Predicates ~~~~~~~~~~~~~~~~~~~~

src/dotdir/runtime/perform.py

@@ -42,18 +42,22 @@ def _rp_perform(
             f"({description!r}) At least one step or argument must"
             f" be supplied to a `perform` function!",
         )
+    code: int
     if _rf_D_is_fn_seq(steps_or_cmd):
-        return _rf_D_perform_fns(*steps_or_cmd)
+        code = _rf_D_perform_fns(*steps_or_cmd)
     elif _rf_D_is_cmd_seq(steps_or_cmd):
         proc = _rf_D_perform_cmds(*steps_or_cmd, pwd=ctx.pwd)
         if (proc.stdout is not None) and (len(proc.stdout) > 0):
             ctx.tfmt.line(BRIGHT_BLACK, proc.stdout, split_on_nl=True)
         if (proc.stderr is not None) and (len(proc.stderr) > 0):
             ctx.tfmt.line(BRIGHT_BLACK, proc.stderr, split_on_nl=True)
-        return proc.returncode
+        code = proc.returncode
     else:
         types = tuple(type(x) for x in steps_or_cmd)
         raise TypeError(f"Invalid types given: {types!r}")
+    ctx.push_code(code)
+    ctx.merge_codes()
+    return code
 
 #  ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 #  ~~~~~~~~~~~~~~~~~~~~ Perform Variants ~~~~~~~~~~~~~~~~~~~~