edq.core.argparser

A place to handle common CLI arguments. "parsers" in this file are always assumed to be argparse parsers.

The general idea is that callers can register callbacks to be called before and after parsing CLI arguments. Pre-callbacks are generally intended to add arguments to the parser, while post-callbacks are generally intended to act on the results of parsing.

  1"""
  2A place to handle common CLI arguments.
  3"parsers" in this file are always assumed to be argparse parsers.
  4
  5The general idea is that callers can register callbacks to be called before and after parsing CLI arguments.
  6Pre-callbacks are generally intended to add arguments to the parser,
  7while post-callbacks are generally intended to act on the results of parsing.
  8"""
  9
 10import argparse
 11import functools
 12import typing
 13
 14import edq.core.config
 15import edq.core.log
 16import edq.util.net
 17
 18@typing.runtime_checkable
 19class PreParseFunction(typing.Protocol):
 20    """
 21    A function that can be called before parsing arguments.
 22    """
 23
 24    def __call__(self, parser: argparse.ArgumentParser, extra_state: typing.Dict[str, typing.Any]) -> None:
 25        """
 26        Prepare a parser for parsing.
 27        This is generally used for adding your module's arguments to the parser,
 28        for example a logging module may add arguments to set a logging level.
 29
 30        The extra state is shared between all pre-parse functions
 31        and will be placed in the final parsed output under `_pre_extra_state_`.
 32        """
 33
 34@typing.runtime_checkable
 35class PostParseFunction(typing.Protocol):
 36    """
 37    A function that can be called after parsing arguments.
 38    """
 39
 40    def __call__(self,
 41            parser: argparse.ArgumentParser,
 42            args: argparse.Namespace,
 43            extra_state: typing.Dict[str, typing.Any]) -> None:
 44        """
 45        Take actions after arguments are parsed.
 46        This is generally used for initializing your module with options,
 47        for example a logging module may set a logging level.
 48
 49        The extra state is shared between all post-parse functions
 50        and will be placed in the final parsed output under `_post_extra_state_`.
 51        """
 52
 53class Parser(argparse.ArgumentParser):
 54    """
 55    Extend an argparse parser to call the pre and post functions.
 56    """
 57
 58    def __init__(self, *args: typing.Any, **kwargs: typing.Any) -> None:
 59        super().__init__(*args, **kwargs)
 60
 61        self._pre_parse_callbacks: typing.Dict[str, PreParseFunction] = {}
 62        self._post_parse_callbacks: typing.Dict[str, PostParseFunction] = {}
 63
 64    def register_callbacks(self,
 65            key: str,
 66            pre_parse_callback: typing.Union[PreParseFunction, None] = None,
 67            post_parse_callback: typing.Union[PostParseFunction, None] = None,
 68            ) -> None:
 69        """
 70        Register callback functions to run before/after argument parsing.
 71        Any existing callbacks under the specified key will be replaced.
 72        """
 73
 74        if (pre_parse_callback is not None):
 75            self._pre_parse_callbacks[key] = pre_parse_callback
 76
 77        if (post_parse_callback is not None):
 78            self._post_parse_callbacks[key] = post_parse_callback
 79
 80    def parse_args(self,  # type: ignore[override]
 81            *args: typing.Any,
 82            skip_keys: typing.Union[typing.List[str], None] = None,
 83            **kwargs: typing.Any) -> argparse.Namespace:
 84        if (skip_keys is None):
 85            skip_keys = []
 86
 87        # Call pre-parse callbacks.
 88        pre_extra_state: typing.Dict[str, typing.Any] = {}
 89        for (key, pre_parse_callback) in self._pre_parse_callbacks.items():
 90            if (key not in skip_keys):
 91                pre_parse_callback(self, pre_extra_state)
 92
 93        # Parse the args.
 94        parsed_args = super().parse_args(*args, **kwargs)
 95
 96        # Call post-parse callbacks.
 97        post_extra_state: typing.Dict[str, typing.Any] = {}
 98        for (key, post_parse_callback) in self._post_parse_callbacks.items():
 99            if (key not in skip_keys):
100                post_parse_callback(self, parsed_args, post_extra_state)
101
102        # Attach the additional state to the args.
103        setattr(parsed_args, '_pre_extra_state_', pre_extra_state)
104        setattr(parsed_args, '_post_extra_state_', post_extra_state)
105
106        return parsed_args  # type: ignore[no-any-return]
107
108def get_default_parser(description: str,
109        version: typing.Union[str, None] = None,
110        include_log: bool = True,
111        include_config: bool = True,
112        include_net: bool = False,
113        config_options: typing.Union[typing.Dict[str, typing.Any], None] = None,
114        ) -> Parser:
115    """ Get a parser with the requested default callbacks already attached. """
116
117    if (config_options is None):
118        config_options = {}
119
120    parser = Parser(description = description)
121
122    if (version is not None):
123        parser.add_argument('--version',
124                action = 'version', version = version)
125
126    if (include_log):
127        parser.register_callbacks('log', edq.core.log.set_cli_args, edq.core.log.init_from_args)
128
129    if (include_config):
130        config_pre_func = functools.partial(edq.core.config.set_cli_args, **config_options)
131        config_post_func = functools.partial(edq.core.config.load_config_into_args, **config_options)
132        parser.register_callbacks('config', config_pre_func, config_post_func)
133
134    if (include_net):
135        parser.register_callbacks('net', edq.util.net.set_cli_args, edq.util.net.init_from_args)
136
137    return parser
@typing.runtime_checkable
class PreParseFunction(typing.Protocol):
19@typing.runtime_checkable
20class PreParseFunction(typing.Protocol):
21    """
22    A function that can be called before parsing arguments.
23    """
24
25    def __call__(self, parser: argparse.ArgumentParser, extra_state: typing.Dict[str, typing.Any]) -> None:
26        """
27        Prepare a parser for parsing.
28        This is generally used for adding your module's arguments to the parser,
29        for example a logging module may add arguments to set a logging level.
30
31        The extra state is shared between all pre-parse functions
32        and will be placed in the final parsed output under `_pre_extra_state_`.
33        """

A function that can be called before parsing arguments.

PreParseFunction(*args, **kwargs)
1953def _no_init_or_replace_init(self, *args, **kwargs):
1954    cls = type(self)
1955
1956    if cls._is_protocol:
1957        raise TypeError('Protocols cannot be instantiated')
1958
1959    # Already using a custom `__init__`. No need to calculate correct
1960    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
1961    if cls.__init__ is not _no_init_or_replace_init:
1962        return
1963
1964    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
1965    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
1966    # searches for a proper new `__init__` in the MRO. The new `__init__`
1967    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
1968    # instantiation of the protocol subclass will thus use the new
1969    # `__init__` and no longer call `_no_init_or_replace_init`.
1970    for base in cls.__mro__:
1971        init = base.__dict__.get('__init__', _no_init_or_replace_init)
1972        if init is not _no_init_or_replace_init:
1973            cls.__init__ = init
1974            break
1975    else:
1976        # should not happen
1977        cls.__init__ = object.__init__
1978
1979    cls.__init__(self, *args, **kwargs)
@typing.runtime_checkable
class PostParseFunction(typing.Protocol):
35@typing.runtime_checkable
36class PostParseFunction(typing.Protocol):
37    """
38    A function that can be called after parsing arguments.
39    """
40
41    def __call__(self,
42            parser: argparse.ArgumentParser,
43            args: argparse.Namespace,
44            extra_state: typing.Dict[str, typing.Any]) -> None:
45        """
46        Take actions after arguments are parsed.
47        This is generally used for initializing your module with options,
48        for example a logging module may set a logging level.
49
50        The extra state is shared between all post-parse functions
51        and will be placed in the final parsed output under `_post_extra_state_`.
52        """

A function that can be called after parsing arguments.

PostParseFunction(*args, **kwargs)
1953def _no_init_or_replace_init(self, *args, **kwargs):
1954    cls = type(self)
1955
1956    if cls._is_protocol:
1957        raise TypeError('Protocols cannot be instantiated')
1958
1959    # Already using a custom `__init__`. No need to calculate correct
1960    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
1961    if cls.__init__ is not _no_init_or_replace_init:
1962        return
1963
1964    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
1965    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
1966    # searches for a proper new `__init__` in the MRO. The new `__init__`
1967    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
1968    # instantiation of the protocol subclass will thus use the new
1969    # `__init__` and no longer call `_no_init_or_replace_init`.
1970    for base in cls.__mro__:
1971        init = base.__dict__.get('__init__', _no_init_or_replace_init)
1972        if init is not _no_init_or_replace_init:
1973            cls.__init__ = init
1974            break
1975    else:
1976        # should not happen
1977        cls.__init__ = object.__init__
1978
1979    cls.__init__(self, *args, **kwargs)
class Parser(argparse.ArgumentParser):
 54class Parser(argparse.ArgumentParser):
 55    """
 56    Extend an argparse parser to call the pre and post functions.
 57    """
 58
 59    def __init__(self, *args: typing.Any, **kwargs: typing.Any) -> None:
 60        super().__init__(*args, **kwargs)
 61
 62        self._pre_parse_callbacks: typing.Dict[str, PreParseFunction] = {}
 63        self._post_parse_callbacks: typing.Dict[str, PostParseFunction] = {}
 64
 65    def register_callbacks(self,
 66            key: str,
 67            pre_parse_callback: typing.Union[PreParseFunction, None] = None,
 68            post_parse_callback: typing.Union[PostParseFunction, None] = None,
 69            ) -> None:
 70        """
 71        Register callback functions to run before/after argument parsing.
 72        Any existing callbacks under the specified key will be replaced.
 73        """
 74
 75        if (pre_parse_callback is not None):
 76            self._pre_parse_callbacks[key] = pre_parse_callback
 77
 78        if (post_parse_callback is not None):
 79            self._post_parse_callbacks[key] = post_parse_callback
 80
 81    def parse_args(self,  # type: ignore[override]
 82            *args: typing.Any,
 83            skip_keys: typing.Union[typing.List[str], None] = None,
 84            **kwargs: typing.Any) -> argparse.Namespace:
 85        if (skip_keys is None):
 86            skip_keys = []
 87
 88        # Call pre-parse callbacks.
 89        pre_extra_state: typing.Dict[str, typing.Any] = {}
 90        for (key, pre_parse_callback) in self._pre_parse_callbacks.items():
 91            if (key not in skip_keys):
 92                pre_parse_callback(self, pre_extra_state)
 93
 94        # Parse the args.
 95        parsed_args = super().parse_args(*args, **kwargs)
 96
 97        # Call post-parse callbacks.
 98        post_extra_state: typing.Dict[str, typing.Any] = {}
 99        for (key, post_parse_callback) in self._post_parse_callbacks.items():
100            if (key not in skip_keys):
101                post_parse_callback(self, parsed_args, post_extra_state)
102
103        # Attach the additional state to the args.
104        setattr(parsed_args, '_pre_extra_state_', pre_extra_state)
105        setattr(parsed_args, '_post_extra_state_', post_extra_state)
106
107        return parsed_args  # type: ignore[no-any-return]

Extend an argparse parser to call the pre and post functions.

Parser(*args: Any, **kwargs: Any)
59    def __init__(self, *args: typing.Any, **kwargs: typing.Any) -> None:
60        super().__init__(*args, **kwargs)
61
62        self._pre_parse_callbacks: typing.Dict[str, PreParseFunction] = {}
63        self._post_parse_callbacks: typing.Dict[str, PostParseFunction] = {}
def register_callbacks( self, key: str, pre_parse_callback: Optional[PreParseFunction] = None, post_parse_callback: Optional[PostParseFunction] = None) -> None:
65    def register_callbacks(self,
66            key: str,
67            pre_parse_callback: typing.Union[PreParseFunction, None] = None,
68            post_parse_callback: typing.Union[PostParseFunction, None] = None,
69            ) -> None:
70        """
71        Register callback functions to run before/after argument parsing.
72        Any existing callbacks under the specified key will be replaced.
73        """
74
75        if (pre_parse_callback is not None):
76            self._pre_parse_callbacks[key] = pre_parse_callback
77
78        if (post_parse_callback is not None):
79            self._post_parse_callbacks[key] = post_parse_callback

Register callback functions to run before/after argument parsing. Any existing callbacks under the specified key will be replaced.

def parse_args( self, *args: Any, skip_keys: Optional[List[str]] = None, **kwargs: Any) -> argparse.Namespace:
 81    def parse_args(self,  # type: ignore[override]
 82            *args: typing.Any,
 83            skip_keys: typing.Union[typing.List[str], None] = None,
 84            **kwargs: typing.Any) -> argparse.Namespace:
 85        if (skip_keys is None):
 86            skip_keys = []
 87
 88        # Call pre-parse callbacks.
 89        pre_extra_state: typing.Dict[str, typing.Any] = {}
 90        for (key, pre_parse_callback) in self._pre_parse_callbacks.items():
 91            if (key not in skip_keys):
 92                pre_parse_callback(self, pre_extra_state)
 93
 94        # Parse the args.
 95        parsed_args = super().parse_args(*args, **kwargs)
 96
 97        # Call post-parse callbacks.
 98        post_extra_state: typing.Dict[str, typing.Any] = {}
 99        for (key, post_parse_callback) in self._post_parse_callbacks.items():
100            if (key not in skip_keys):
101                post_parse_callback(self, parsed_args, post_extra_state)
102
103        # Attach the additional state to the args.
104        setattr(parsed_args, '_pre_extra_state_', pre_extra_state)
105        setattr(parsed_args, '_post_extra_state_', post_extra_state)
106
107        return parsed_args  # type: ignore[no-any-return]
def get_default_parser( description: str, version: Optional[str] = None, include_log: bool = True, include_config: bool = True, include_net: bool = False, config_options: Optional[Dict[str, Any]] = None) -> Parser:
109def get_default_parser(description: str,
110        version: typing.Union[str, None] = None,
111        include_log: bool = True,
112        include_config: bool = True,
113        include_net: bool = False,
114        config_options: typing.Union[typing.Dict[str, typing.Any], None] = None,
115        ) -> Parser:
116    """ Get a parser with the requested default callbacks already attached. """
117
118    if (config_options is None):
119        config_options = {}
120
121    parser = Parser(description = description)
122
123    if (version is not None):
124        parser.add_argument('--version',
125                action = 'version', version = version)
126
127    if (include_log):
128        parser.register_callbacks('log', edq.core.log.set_cli_args, edq.core.log.init_from_args)
129
130    if (include_config):
131        config_pre_func = functools.partial(edq.core.config.set_cli_args, **config_options)
132        config_post_func = functools.partial(edq.core.config.load_config_into_args, **config_options)
133        parser.register_callbacks('config', config_pre_func, config_post_func)
134
135    if (include_net):
136        parser.register_callbacks('net', edq.util.net.set_cli_args, edq.util.net.init_from_args)
137
138    return parser

Get a parser with the requested default callbacks already attached.