runtime.exec

Contains all of the middlewares used during execution of a command

ApplyParseResultMiddleware

Bases: ParamProcessor

Update the parameter values with the values found from parsing the input

Context Dependencies

• arc.parse.result
• arc.args.tree
• arc.config
• arc.args.origins

Context Additions

None

Source code in /home/runner/work/arc/arc/arc/runtime/exec.py

class ApplyParseResultMiddleware(ParamProcessor):
    """Update the parameter values with the values found from parsing the input

    # Context Dependencies
    - `arc.parse.result`
    - `arc.args.tree`
    - `arc.config`
    - `arc.args.origins`

    # Context Additions
    None
    """

    res: at.ParseResult

    def __call__(self, ctx: Context) -> t.Any:
        self.res = ctx["arc.parse.result"]
        return super().__call__(ctx)

    def process_missing(self, param: Param[t.Any]) -> t.Any:
        value: t.Any = self.res.pop(param.argument_name, constants.MISSING)
        self.set_origin(param, ValueOrigin.COMMAND_LINE)

        # NOTE: This is dependent on the fact that the current parser
        # will return None for append action type parameters that
        # haven't been given any values.
        if value is None:
            if param.is_required:
                raise errors.MissingOptionValueError(param)
            else:
                return constants.MISSING

        return value

CompileParamsMiddleware

Bases: MiddlewareBase

Compile the parameter tree into dictionary that can be used to call the command callback.

Context Dependencies

• arc.args.tree

Context Additions

• arc.args - Dictionary to pass into command callback

Source code in /home/runner/work/arc/arc/arc/runtime/exec.py

class CompileParamsMiddleware(MiddlewareBase):
    """Compile the parameter tree into dictionary that can be used
    to call the command callback.

    # Context Dependencies
    - `arc.args.tree`

    # Context Additions
    - `arc.args` - Dictionary to pass into command callback

    """

    def __call__(self, ctx: Context) -> t.Any:
        instance: ParamInstanceTree[t.Any] = ctx["arc.args.tree"]
        ctx["arc.args"] = instance.compile()

ConvertValuesMiddleware

Bases: ParamProcessor

Performs type conversion on retrieved parameter values

Context Dependencies

• arc.args.tree
• arc.config
• arc.args.origins

Context Additions

None

Source code in /home/runner/work/arc/arc/arc/runtime/exec.py

class ConvertValuesMiddleware(ParamProcessor):
    """Performs type conversion on retrieved parameter values

    # Context Dependencies
    - `arc.args.tree`
    - `arc.config`
    - `arc.args.origins`

    # Context Additions
    None
    """

    def process_not_missing(self, param: Param[t.Any], value: t.Any) -> t.Any:
        if value in (None, constants.MISSING, True, False) and param.type.origin in (
            bool,
            t.Any,
        ):
            return value

        try:
            return param.convert(value)
        except errors.ConversionError as e:
            details = e.details

            if self.ctx.get_origin(param.argument_name) == ValueOrigin.ENV:
                details = f"Invalid value in enviroment variable: {self.config.env_prefix}{param.envvar}"

            raise errors.InvalidParamValueError(str(e), param, details) from e

DefaultValueMiddleware

Bases: ParamProcessor

Retrieves parameter values from the defaults for each parameter

Context Dependencies

• arc.args.tree
• arc.config
• arc.args.origins

Context Additions

None

Source code in /home/runner/work/arc/arc/arc/runtime/exec.py

class DefaultValueMiddleware(ParamProcessor):
    """Retrieves parameter values from the defaults for each parameter

    # Context Dependencies
    - `arc.args.tree`
    - `arc.config`
    - `arc.args.origins`

    # Context Additions
    None
    """

    def process_missing(self, param: Param[t.Any]) -> t.Any:
        self.set_origin(param, ValueOrigin.DEFAULT)
        return param.default

DependancyInjectorMiddleware

Bases: ParamProcessor

Performs dependcy injection for relavent parameters

Context Dependencies

• arc.args.tree
• arc.config
• arc.args.origins

Context Additions

None

Source code in /home/runner/work/arc/arc/arc/runtime/exec.py

class DependancyInjectorMiddleware(ParamProcessor):
    """Performs dependcy injection for relavent parameters

    # Context Dependencies
    - `arc.args.tree`
    - `arc.config`
    - `arc.args.origins`

    # Context Additions
    None
    """

    def process(self, param: Param[t.Any], value: t.Any) -> t.Any:
        param = t.cast(InjectedParam[t.Any], param)
        injected = param.get_injected_value(self.ctx)
        self.set_origin(param, ValueOrigin.INJECTED)
        return injected

    def skip(self, param: Param[t.Any], _value: t.Any) -> bool:
        return not param.is_injected

ExecMiddleware

Bases: DefaultMiddlewareNamespace

Container for all default execution middlewares

Use it to reference a default exec middleware when adding your own custom middlewares

import arc

@arc.command
def command():
    arc.print("hello there")

@command.use(after=arc.ExecMiddleware.ConverValues)
def after_convert(ctx: arc.Context):
    # Runs after type conversion occurs
    ...

Source code in /home/runner/work/arc/arc/arc/runtime/exec.py

class ExecMiddleware(DefaultMiddlewareNamespace):
    """Container for all default execution middlewares

    Use it to reference a default exec middleware when adding your own custom middlewares

    ```py
    import arc

    @arc.command
    def command():
        arc.print("hello there")

    @command.use(after=arc.ExecMiddleware.ConverValues)
    def after_convert(ctx: arc.Context):
        # Runs after type conversion occurs
        ...
    ```
    """

    ExitStack = ExitStackMiddleware()
    SetupParam = SetupParamMiddleware()
    ApplyParseResult = ApplyParseResultMiddleware()
    GetEnvValue = GetEnvValueMiddleware()
    GetPromptValue = GetPromptValueMiddleware()
    GetterValue = GetterValueMiddleware()
    ConvertValues = ConvertValuesMiddleware()
    DefaultValue = DefaultValueMiddleware()
    DependancyInjector = DependancyInjectorMiddleware()
    RunTypeMiddleware = RunTypeMiddlewareMiddleware()
    RunCallbacks = RunCallbacksMiddleware()
    MissingParamsChecker = MissingParamsCheckerMiddleware()
    CompileParams = CompileParamsMiddleware()
    OpenResource = OpenResourceMiddleware()

    _list: list[Middleware] = [
        ExitStack,
        SetupParam,
        ApplyParseResult,
        GetEnvValue,
        GetPromptValue,
        GetterValue,
        ConvertValues,
        DefaultValue,
        DependancyInjector,
        RunTypeMiddleware,
        RunCallbacks,
        MissingParamsChecker,
        CompileParams,
        OpenResource,
    ]

ExitStackMiddleware

Bases: MiddlewareBase

Utility middleware that adds an instance of contextlib.ExitStack to the context. This can be used to open resources (like IO objects) that need to be closed when the program is exiting.

Context Dependencies

None

Context Additions

• arc.exitstack

Source code in /home/runner/work/arc/arc/arc/runtime/exec.py

class ExitStackMiddleware(MiddlewareBase):
    """Utility middleware that adds an instance of `contextlib.ExitStack` to the context.
    This can be used to open resources (like IO objects) that need to be closed when the program is exiting.

    # Context Dependencies
    None

    # Context Additions
    - `arc.exitstack`
    """

    def __call__(self, ctx: Context) -> t.Any:
        with contextlib.ExitStack() as stack:
            ctx["arc.exitstack"] = stack
            yield
            if length := len(stack._exit_callbacks):  # type: ignore
                ctx.logger.debug("Closing %d resource(s)", length)

GetEnvValueMiddleware

Bases: ParamProcessor

Retrieves parameter values from enviroment variables

Context Dependencies

• arc.args.tree
• arc.config
• arc.args.origins

Context Additions

None

Source code in /home/runner/work/arc/arc/arc/runtime/exec.py

class GetEnvValueMiddleware(ParamProcessor):
    """Retrieves parameter values from enviroment variables

    # Context Dependencies
    - `arc.args.tree`
    - `arc.config`
    - `arc.args.origins`

    # Context Additions
    None
    """

    def process_missing(self, param: Param[t.Any]) -> t.Any:
        value = self.get_env_value(param)
        self.set_origin(param, ValueOrigin.ENV)
        return value

    def get_env_value(self, param: Param[t.Any]) -> str | constants.Constant:
        if not param.envvar:
            return constants.MISSING

        return os.getenv(f"{self.config.env_prefix}{param.envvar}", constants.MISSING)

GetPromptValueMiddleware

Bases: ParamProcessor

Retrieves parameter values from user input

Context Dependencies

• arc.args.tree
• arc.config
• arc.args.origins

Context Additions

None

Source code in /home/runner/work/arc/arc/arc/runtime/exec.py

class GetPromptValueMiddleware(ParamProcessor):
    """Retrieves parameter values from user input

    # Context Dependencies
    - `arc.args.tree`
    - `arc.config`
    - `arc.args.origins`

    # Context Additions
    None
    """

    def process_missing(self, param: Param[t.Any]) -> t.Any:
        value = self.get_prompt_value(param)
        self.set_origin(param, ValueOrigin.PROMPT)
        return value

    def get_prompt_value(self, param: Param[t.Any]) -> str | constants.Constant:
        if not param.prompt:
            return constants.MISSING

        prompter = getattr(param.type.resolved_type, "__prompt__", input_prompt)
        return api.dispatch_args(prompter, param, self.ctx)

GetterValueMiddleware

Bases: ParamProcessor

Retrieves parameter values from the parameter getter function

Context Dependencies

• arc.args.tree
• arc.config
• arc.args.origins

Context Additions

None

Source code in /home/runner/work/arc/arc/arc/runtime/exec.py

class GetterValueMiddleware(ParamProcessor):
    """Retrieves parameter values from the parameter getter function

    # Context Dependencies
    - `arc.args.tree`
    - `arc.config`
    - `arc.args.origins`

    # Context Additions
    None
    """

    def process_missing(self, param: Param[t.Any]) -> t.Any:
        value = self.get_getter_value(param)
        self.set_origin(param, ValueOrigin.GETTER)
        return value

    def get_getter_value(self, param: Param[t.Any]) -> t.Any | constants.Constant:
        getter = param.getter_func
        if not getter:
            return constants.MISSING

        return api.dispatch_args(getter, param, self.ctx)

MissingParamsCheckerMiddleware

Bases: MiddlewareBase

Checks to ensure that all params have been given a value

Context Dependencies

• arc.args.tree

Context Additions

None

Source code in /home/runner/work/arc/arc/arc/runtime/exec.py

class MissingParamsCheckerMiddleware(MiddlewareBase):
    """Checks to ensure that all params have been given a value

    # Context Dependencies
    - `arc.args.tree`

    # Context Additions
    None
    """

    def __call__(self, ctx: Context) -> t.Any:
        tree: ParamInstanceTree[t.Any] = ctx["arc.args.tree"]
        missing = [
            value.param
            for value in tree.leaves()
            if value.value is constants.MISSING and value.param.expose
        ]

        if missing:
            raise arc.errors.MissingArgValueError(missing)

OpenResourceMiddleware

Bases: MiddlewareBase

Opens any parameter that is a context manager

Context Dependencies

• arc.args
• arc.exitstack

Context Additions

None

Source code in /home/runner/work/arc/arc/arc/runtime/exec.py

class OpenResourceMiddleware(MiddlewareBase):
    """Opens any parameter that is a context manager

    # Context Dependencies
    - `arc.args`
    - `arc.exitstack`

    # Context Additions
    None
    """

    def __call__(self, ctx: Context) -> t.Any:
        stack: contextlib.ExitStack | None = ctx.get("arc.exitstack")
        if not stack:
            return

        args: dict[str, t.Any] = ctx["arc.args"]

        for key, val in args.items():
            if isinstance(val, constants.COLLECTION_TYPES):
                cls = type(val)
                args[key] = cls(
                    item if not iscontextmanager(item) else stack.enter_context(item)
                    for item in val
                )

            if iscontextmanager(val):
                args[key] = stack.enter_context(val)

RunTypeMiddlewareMiddleware

Bases: ParamProcessor

Execute the registered type middleware for each parameter

Context Dependencies

• arc.args.tree
• arc.config
• arc.args.origins

Context Additions

None

Source code in /home/runner/work/arc/arc/arc/runtime/exec.py

class RunTypeMiddlewareMiddleware(ParamProcessor):
    """Execute the registered type middleware for each parameter

    # Context Dependencies
    - `arc.args.tree`
    - `arc.config`
    - `arc.args.origins`

    # Context Additions
    None
    """

    def process(self, param: Param[t.Any], value: t.Any) -> t.Any:
        if value not in (None, constants.MISSING):
            try:
                value = param.run_middleware(value, self.ctx)
            except errors.ValidationError as e:
                raise errors.InvalidParamValueError(str(e), param) from e
        return value

    def skip(self, _param: Param[t.Any], _value: t.Any) -> bool:
        return False  # Don't want to skip any of the params

SetupParamMiddleware

Bases: MiddlewareBase

Performs parameter setup for the given command

Context Dependencies

• arc.command

Context Additions

arc.args.tree - Tree representing the command's parameters and their realized values arc.args.origins - Dictionary that stores where each parameter value comes from. See Context.get_origin()

Source code in /home/runner/work/arc/arc/arc/runtime/exec.py

class SetupParamMiddleware(MiddlewareBase):
    """Performs parameter setup for the given command

    # Context Dependencies
    - `arc.command`

    # Context Additions
    `arc.args.tree` - Tree representing the command's parameters and their realized values
    `arc.args.origins` - Dictionary that stores where each parameter value comes from. See `Context.get_origin()`
    """

    def __call__(self, ctx: Context) -> t.Any:
        command: Command = ctx["arc.command"]
        param_instance = command.param_def.create_instance()
        ctx["arc.args.tree"] = param_instance
        ctx.setdefault("arc.args.origins", {})
        ctx.logger.debug("Parsed input: %s", ctx.get("arc.parse.result"))