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
PreParseFunction19@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
PostParseFunction35@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.
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.