Skip to content

context ¤

Module containing the context definition.

Classes:

  • Context –

    A simple context class.

Context ¤

Context(
    options: dict[str, Any],
    options_override: dict[str, Any] | None = None,
)

A simple context class.

Context instances are passed to functions decorated with duty.

Parameters:

  • options (dict[str, Any]) –

    Base options specified in @duty(**options).

  • options_override (dict[str, Any] | None, default: None ) –

    Options that override run and @duty options. This argument is used to allow users to override options from the CLI or environment.

Methods:

  • cd –

    Change working directory as a context manager.

  • options –

    Change options as a context manager.

  • run –

    Run a command in a subprocess or a Python callable.

Source code in src/duty/context.py 
22
23
24
25
26
27
28
29
30
31
32
def __init__(self, options: dict[str, Any], options_override: dict[str, Any] | None = None) -> None:
    """Initialize the context.

    Parameters:
        options: Base options specified in `@duty(**options)`.
        options_override: Options that override `run` and `@duty` options.
            This argument is used to allow users to override options from the CLI or environment.
    """
    self._options = options
    self._option_stack: list[dict[str, Any]] = []
    self._options_override = options_override or {}

cd ¤

cd(directory: str) -> Iterator

Change working directory as a context manager.

Parameters:

  • directory (str) –

    The directory to go into.

Yields:

Source code in src/duty/context.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@contextmanager
def cd(self, directory: str) -> Iterator:
    """Change working directory as a context manager.

    Parameters:
        directory: The directory to go into.

    Yields:
        Nothing.
    """
    if not directory:
        yield
        return
    old_wd = os.getcwd()
    os.chdir(directory)
    try:
        yield
    finally:
        os.chdir(old_wd)

options ¤

options(**opts: Any) -> Iterator

Change options as a context manager.

Can be nested as will, previous options will pop once out of the with clause.

Parameters:

  • **opts (Any, default: {} ) –

    Options used in run.

Yields:

Source code in src/duty/context.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@contextmanager
def options(self, **opts: Any) -> Iterator:
    """Change options as a context manager.

    Can be nested as will, previous options will pop once out of the with clause.

    Parameters:
        **opts: Options used in `run`.

    Yields:
        Nothing.
    """
    self._option_stack.append(self._options)
    self._options = {**self._options, **opts}
    try:
        yield
    finally:
        self._options = self._option_stack.pop()

run ¤

run(cmd: CmdType, **options: Any) -> str

Run a command in a subprocess or a Python callable.

Parameters:

  • cmd (CmdType) –

    A command or a Python callable.

  • options (Any, default: {} ) –

    Options passed to failprint functions.

Raises:

  • DutyFailure –

    When the exit code / function result is greather than 0.

Returns:

  • str –

    The output of the command.

Source code in src/duty/context.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def run(self, cmd: CmdType, **options: Any) -> str:
    """Run a command in a subprocess or a Python callable.

    Parameters:
        cmd: A command or a Python callable.
        options: Options passed to `failprint` functions.

    Raises:
        DutyFailure: When the exit code / function result is greather than 0.

    Returns:
        The output of the command.
    """
    final_options = dict(self._options)
    final_options.update(options)

    allow_overrides = final_options.pop("allow_overrides", True)
    workdir = final_options.pop("workdir", None)

    if allow_overrides:
        final_options.update(self._options_override)

    with self.cd(workdir):
        try:
            result = failprint_run(cmd, **final_options)
        except KeyboardInterrupt as ki:
            raise DutyFailure(130) from ki

    if result.code:
        raise DutyFailure(result.code)

    return result.output