| Viewing file:  argparse_utils.py (11.25 KB)      -rw-r--r-- Select action/file-type:
 
  (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) | 
 
# coding=utf-8#
 # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
 #
 # Licensed under CLOUD LINUX LICENSE AGREEMENT
 # http://cloudlinux.com/docs/LICENSE.TXT
 
 """
 Extensions for validation and output formatting argparse
 """
 
 import argparse
 import pwd
 import datetime
 
 from prettytable import PrettyTable
 
 from lvestats.lib.commons.dateutil import parse_date, parse_period2
 
 
 class SmartFormatter(argparse.HelpFormatter):
 """
 Format argparse help in many line, to format many lines use "R|" at start line
 >>>import argparse
 >>>parser = argparse.ArgumentParser(formatter_class=SmartFormatter)
 >>>parser.add_argument('-a', '--argument', help="R| Descriptions: \\n first new line\\n second new line")
 """
 def _split_lines(self, text, width):
 if text.startswith('R|'):
 return text[2:].splitlines()
 return argparse.HelpFormatter._split_lines(self, text, width)  # pylint: disable=protected-access
 
 def _format_args(self, action, default_metavar):
 if isinstance(action, ParseDatetime):
 get_metavar = self._metavar_formatter(action, default_metavar)
 return str(get_metavar(1))
 return argparse.HelpFormatter._format_args(self, action, default_metavar)  # pylint: disable=protected-access
 
 
 class ParseDatetime(argparse.Action):
 """
 Interprets a string ([YY]YY-MM-DD[ HH:MM]) as a datetime for argument parsing.
 return parsed data as <type 'datetime.datetime'>
 """
 def __call__(self, parser, namespace, values, option_string=None):
 datetime_str = ' '.join(values) if hasattr(values, '__iter__') else values
 try:
 result_datetime = parse_date(datetime_str)
 except ValueError as e:
 raise argparse.ArgumentError(
 self,
 f"{datetime_str} is an invalid; use [YY]YY-MM-DD[ HH:MM] format",
 ) from e
 setattr(namespace, self.dest, result_datetime)
 
 
 def check_percentages(value):
 """
 Check if input argument between 0 and 100
 :param str value:
 :return int: percent
 :except argparse.ArgumentTypeError: generate exception for command line warning message
 """
 value = int(value)
 if value < 0 or value > 100:
 raise argparse.ArgumentTypeError(f"{value} is an invalid; this value can range from 0 to 100")
 return value
 
 
 def check_non_negative_int(value):
 """
 Check if input argument greater than (or equal to) zero and not float
 :param str value:
 :return int:
 :except argparse.ArgumentTypeError: generate exception for command line warning message
 """
 try:
 non_negative_int = int(value)
 except ValueError as e:
 raise argparse.ArgumentTypeError(
 f'"{value}" is an invalid; must be a number greater than (or equal to) zero'
 ) from e
 if '.' in value or non_negative_int < 0:
 raise argparse.ArgumentTypeError(
 f'"{value}" is an invalid; must be a number greater than (or equal to) zero'
 )
 return non_negative_int
 
 
 def check_positive_int(value):
 """
 Check if input argument is digest, greater than zero and not float
 :param str value: line to check
 :return int: positive integer
 :except argparse.ArgumentTypeError: generate exception for command line warning message
 """
 try:
 positive_int = int(value)
 except ValueError as e:
 raise argparse.ArgumentTypeError(
 f'"{value}" is an invalid; must be a number greater than zero'
 ) from e
 if '.' in value or positive_int <= 0:
 raise argparse.ArgumentTypeError(
 f'"{value}" is an invalid; must be a number greater than zero'
 )
 return positive_int
 
 
 def check_timestamp(value):
 """
 Check if input argument is a correct timestamp
 253370754000.0 == time.mktime(datetime.datetime(year=datetime.MAXYEAR, month=1, day=1).timetuple())
 :param str value: line to check
 :return float: timestamp
 :except argparse.ArgumentTypeError: generate exception for command line warning message
 """
 try:
 positive_float = float(value)
 except ValueError as e:
 raise argparse.ArgumentTypeError(
 f'"{value}" is an invalid; timestamp must be a number between 0 and 253370754000'
 ) from e
 if not 0 < positive_float < 253370754000:
 raise argparse.ArgumentTypeError(
 f'"{value}" is an invalid; timestamp must be a number between 0 and 253370754000'
 )
 return positive_float
 
 
 def check_from_zero_to_max_int(value):
 """
 Check if input argument less then maximum integer number
 PostgreSQL - 2147483647
 MySQL - 2147483647
 SQLite - 9223372036854775807
 :param str value:
 :return int:
 :except argparse.ArgumentTypeError: generate exception for command line warning message
 """
 max_int = 2147483647
 try:
 non_negative_int = int(value)
 except ValueError as e:
 raise argparse.ArgumentTypeError(
 f'"{value}" is an invalid; must be a number greater than (or equal to) zero '
 f'and less than {max_int}'
 ) from e
 if '.' in value or non_negative_int < 0 or non_negative_int > max_int:
 raise argparse.ArgumentTypeError(
 f'"{value}" is an invalid; must be a number greater than (or equal to) zero '
 f'and less than {max_int}'
 )
 return non_negative_int
 
 
 def check_user_name(value):
 """
 Check if user present in system and convert user name to user id
 :param str value: user name (login)
 :return int: user id
 :except argparse.ArgumentTypeError: generate exception for command line warning message
 """
 try:
 user_id = pwd.getpwnam(value).pw_uid
 except KeyError as e:
 raise argparse.ArgumentTypeError(f'user "{value}" not present in system') from e
 return user_id
 
 
 def period_type(value):
 """
 Parse multiple period line, and validate input data
 :param str value: line to parsing
 :return int: period in seconds
 :except argparse.ArgumentTypeError: generate exception for command line warning message
 """
 try:
 int_converted = int(value)
 if '.' in value or int_converted <= 0:
 raise argparse.ArgumentTypeError(
 f'"{value}" is an invalid; time period should be greater than (or equal to) '
 '1 minute (60 seconds); not negative'
 )
 value = str(int_converted) + 'm'  # if only digest convert to minutes
 except ValueError:
 pass
 
 try:
 seconds = parse_period2(value)
 except ValueError as e:
 raise argparse.ArgumentTypeError(str(e)) from e
 if seconds < 60:
 raise argparse.ArgumentTypeError(
 f'"{value}" is an invalid; time period should be greater than (or equal to) '
 '1 minute (60 seconds)'
 )
 return seconds
 
 
 def period_type2(value, datetime_now=None, round_to_minutes=True):
 """
 Extented period_type function
 Added support 'yesterday' and 'today'
 :param bool round_to_minutes:
 :param str value: line to parsing
 :return tuple: period from to (<type 'datetime.datetime'>, <type 'datetime.datetime'>)
 :except argparse.ArgumentTypeError: generate exception for command line warning message
 """
 datetime_now = datetime_now or datetime.datetime.utcnow()
 value = value.lower()
 if value == 'yesterday':
 datetime_to = datetime_now.replace(hour=0, minute=0, second=0, microsecond=0)  # today midnight
 datetime_from = datetime_to - datetime.timedelta(days=1)  # yesterday midnight
 elif value == 'today':
 datetime_from = datetime_now.replace(hour=0, minute=0, second=0, microsecond=0)  # today midnight
 datetime_to = datetime_now
 else:
 if round_to_minutes:
 datetime_from = datetime_now.replace(second=0, microsecond=0)  # rounded to minutes
 datetime_to = datetime_now.replace(second=0, microsecond=0)  # rounded to minutes
 else:
 datetime_from = datetime_now
 datetime_to = datetime_now
 datetime_from -= datetime.timedelta(seconds=period_type(value))
 return datetime_from, datetime_to
 
 
 def format_aliases(description):
 """
 Prepare aliases descriptions to table formatted
 :param tuple|list|iter description: massive of data
 :return str: ascii table
 """
 fields = description[0]
 description = description[1:]
 table = PrettyTable(fields, padding_width=0, left_padding=2)
 for field in fields:
 table.align[field] = "l"
 list(map(table.add_row, description))
 return table.get_string(border=False)
 
 
 BY_USAGE_PERCENT = 90
 
 
 def alias_action(alias_dscr):
 alias_dscr = alias_dscr[1:]  # first line is headers for descriptions in print tables
 aliases_func_dict = _alias_dscr_to_dict(alias_dscr)
 
 class AliasAction(argparse.Action):
 def __call__(self, parser, namespace, values, option_string=None):
 
 output_as_list = self.nargs in ('+', '*')
 
 if output_as_list:
 values_ = []
 for val_ in values:  # for using ',' as separator in options
 values_.extend(filter(bool, val_.lower().split(',')))
 # "filter(bool" for correct processing --show-columns=,ID,
 else:
 values_ = [values.strip('"').strip("'").lower()]
 
 # generate exception for not supported aliases
 not_supported_values = set(values_).difference(set(aliases_func_dict))
 if not_supported_values:
 supported_values = tuple(aliases_func_dict)
 unsupported_aliases = '", "'.join(not_supported_values)
 raise argparse.ArgumentError(
 self,
 f'alias(es) "{unsupported_aliases}" not valid; '
 f'choice from: {", ".join(supported_values)}'
 )
 
 result = [aliases_func_dict[_] for _ in values_]  # replace aliases to main aliases
 if not output_as_list:
 result = result[0]
 setattr(namespace, self.dest, result)  # write result to namespace
 return AliasAction
 
 
 def _alias_dscr_to_dict(alias_dscr):
 """
 preparation of the dictionary to extract the names of the aggregation function name (used in show_history)
 as aliases are used from the first to the penultimate elements alias_dscr
 as aggregation function name are used penultimate element
 
 >>> _alias_dscr_to_dict((('al_1', 'al_2', 'al_m1', 'description1'), ('al_3', 'al_4', 'al_m2', 'description1')))
 {'al_2': 'al_m1', 'al_1': 'al_m1', 'al_m1': 'al_m1', 'al_3': 'al_m2', 'al_m2': 'al_m2', 'al_4': 'al_m2'}
 :param tuple|list alias_dscr:
 :type return: dict
 """
 val_index_ = -2
 aliases_func_pairs = [(_[val_index_].lower(), _[val_index_])for _ in alias_dscr]
 for key_index_ in range(len(alias_dscr[0]) - 1):
 aliases_func_pairs.extend([(_[key_index_], _[val_index_]) for _ in alias_dscr])
 aliases_func_dict = dict(aliases_func_pairs)
 return aliases_func_dict
 
 
 def argparse_process_period(namespace):
 # prepare period
 period_from, period_to = getattr(namespace, 'from'), namespace.to
 if namespace.period:
 period_from, period_to = namespace.period
 setattr(namespace, 'from', period_from)
 namespace.to = period_to
 return namespace
 |