Parse timeout input to return instance of timedelta timeout_input(str): String to parse for one or more integer values followed by time unit names. The allowed time units are found in the list constants.TIMEOUT_ALLOWED_UNITS. Any substring from the beginning of the allowed unit
(timeout_input)
| 8 | |
| 9 | |
| 10 | def parse_timeout(timeout_input): |
| 11 | """Parse timeout input to return instance of timedelta |
| 12 | |
| 13 | timeout_input(str): String to parse for one or more integer values followed by time unit names. The allowed time |
| 14 | units are found in the list constants.TIMEOUT_ALLOWED_UNITS. Any substring from the beginning of the allowed |
| 15 | unit may be provided. If no time unit names are included, then the default time unit from |
| 16 | constants.TIMEOUT_DEFAULT_UNIT is used. An error is raised if a unit name in timeout_input is unrecognized. |
| 17 | Returns instance of timedelta from the unit values and names. |
| 18 | """ |
| 19 | if timeout_input.strip().isnumeric(): |
| 20 | tdelta_kwargs = {TIMEOUT_DEFAULT_UNIT: int(timeout_input)} |
| 21 | else: |
| 22 | all_units = TIMEOUT_ALLOWED_UNITS |
| 23 | tdelta_kwargs = {} |
| 24 | for v, input_unit in findall(r'(\d+)\s*([a-zA-Z]+)\s*', timeout_input): |
| 25 | input_unit = input_unit.lower() |
| 26 | key_match = [t[0] for t in all_units if input_unit == t[0] or input_unit == t[1]] |
| 27 | if len(key_match) == 0: |
| 28 | supported_units = ', '.join((f'{x[0]}/{x[1]}' for x in TIMEOUT_ALLOWED_UNITS)) |
| 29 | raise ValueError( |
| 30 | f'{input_unit} is not allowed as a unit for the timeout value.\n' |
| 31 | f'Valid units are "{supported_units}".' |
| 32 | ) |
| 33 | tdelta_kwargs[key_match[0]] = int(v) |
| 34 | days = tdelta_kwargs.get('days') or 0 |
| 35 | if 'years' in tdelta_kwargs: |
| 36 | years = tdelta_kwargs.pop('years') |
| 37 | days += years * 365 |
| 38 | if 'months' in tdelta_kwargs: |
| 39 | months = tdelta_kwargs.pop('months') |
| 40 | days += months * 30 |
| 41 | if days > 0: |
| 42 | tdelta_kwargs['days'] = days |
| 43 | return timedelta(**tdelta_kwargs) |
| 44 | |
| 45 | |
| 46 | def format_timeout(timeout_delta): |