'''
@author: Tsuyoshi Hombashi
'''
import logging
import six
import thutils.common as common
from thutils.logger import logger
TN_ExecuteOption = "Execute Option"
[docs]class MakeOption:
MAKE = "make"
OVERWRITE = "overwrite"
CLEAN = "clean"
SKIP = "skip"
[docs]class ArgumentParserObject(object):
"""
wrapper class of argparse
"""
[docs] class GroupName:
MISC = "Miscellaneous"
SQL = "SQL"
TIME_RANGE = "Time Range"
PROFILE = "Profile"
def __init__(self):
self.parser = None
self.dict_group = {}
[docs] def make(self, version, description="", epilog=""):
import argparse
self.parser = argparse.ArgumentParser(
formatter_class=argparse.RawDescriptionHelpFormatter,
description=description, epilog=epilog)
self.parser.add_argument(
'--version', action='version', version='%(prog)s ' + version)
self._add_general_argument_group()
self._add_log_level_argument_group()
[docs] def parse_args(self):
return self.parser.parse_args()
[docs] def add_argument_group(self, group_name):
if common.is_empty_string(group_name):
raise ValueError("null argument group name")
if group_name not in self.dict_group:
group = self.parser.add_argument_group(group_name)
self.dict_group[group_name] = group
return group
[docs] def add_dry_run_option(self):
group = self.dict_group.get(self.GroupName.MISC)
group.add_argument(
"--dry-run", action="store_true", default=False,
help="do no harm.")
[docs] def add_run_option(self):
group = self.dict_group.get(self.GroupName.MISC)
group.add_argument(
"--run", dest="dry_run", action="store_false", default=True,
help="execute")
[docs] def add_sql_argument_group(self):
group = self.add_argument_group(self.GroupName.SQL)
group.add_argument(
"--sql-logging", action="store_true", default=False,
help="for debug")
return group
[docs] def add_profile_argument_group(self):
group = self.add_argument_group(self.GroupName.PROFILE)
group.add_argument(
"--profile", action="store_true", default=False,
help="for execution time profile (python 2.6 or greater required)")
return group
[docs] def add_time_argument_group(self):
group = self.add_argument_group(self.GroupName.PROFILE)
group.add_argument(
"--time-measure", action="store_const",
const=logging.INFO, default=logging.DEBUG,
help="measuring execution time.")
return group
[docs] def add_time_range_argument_group(
self, valid_time_format_list,
start_time_help_msg="", end_time_help_msg=""):
import re
if common.is_empty_list_or_tuple(valid_time_format_list):
raise ValueError("required at least a valid time format")
# convert datetime format to human readable text
help_time_format_list = []
for time_format in valid_time_format_list:
time_format = re.sub(re.escape("%Y"), "YYYY", time_format)
time_format = re.sub(re.escape("%y"), "YY", time_format)
time_format = re.sub(re.escape("%m"), "MM", time_format)
time_format = re.sub(re.escape("%d"), "DD", time_format)
time_format = re.sub(re.escape("%H"), "hh", time_format)
time_format = re.sub(re.escape("%M"), "mm", time_format)
time_format = re.sub(re.escape("%S"), "ss", time_format)
help_time_format_list.append(time_format)
help_text_format = "%s (valid time format: %s)"
group = self.add_argument_group(self.GroupName.TIME_RANGE)
group.add_argument(
"-s", dest="start_time",
help=help_text_format % (
start_time_help_msg, " | ".join(help_time_format_list)))
group.add_argument(
"-e", dest="end_time",
help=help_text_format % (
end_time_help_msg, " | ".join(help_time_format_list)))
return group
[docs] def addMakeArgumentGroup(self):
dest = "make_option"
group = self.parser.add_mutually_exclusive_group()
group.add_argument(
"--clean", dest=dest, action="store_const",
const=MakeOption.CLEAN, default=MakeOption.MAKE,
help="remove existing file")
group.add_argument(
"--overwrite", dest=dest, action="store_const",
const=MakeOption.OVERWRITE, default=MakeOption.MAKE,
help="overwrite existing file")
@staticmethod
[docs] def validate_time_range_option(
options, valid_time_format_list,
is_check_time_inversion=True):
import thutils.gtime as gtime
try:
options.start_datetime = gtime.toDateTimeEx(
options.start_time, valid_time_format_list)
options.end_datetime = gtime.toDateTimeEx(
options.end_time, valid_time_format_list)
except ValueError:
return
options.datetime_range = gtime.DateTimeRange(
options.start_datetime, options.end_datetime)
if not is_check_time_inversion:
return
try:
options.datetime_range.verifyTimeRange()
except TypeError:
pass
def _add_general_argument_group(self):
group = self.add_argument_group(self.GroupName.MISC)
group.add_argument(
"--logging", dest="with_no_log", action="store_false", default=True,
help="output execution log to a file (%(prog)s.log).")
group.add_argument(
"--stacktrace", action="store_true", default=False,
help="display stack trace when an error occurred.")
return group
def _add_log_level_argument_group(self):
dest = "log_level"
group = self.parser.add_mutually_exclusive_group()
group.add_argument(
"--debug", dest=dest, action="store_const",
const=logging.DEBUG, default=logging.INFO,
help="for debug print.")
group.add_argument(
"--quiet", dest=dest, action="store_const",
const=logging.NOTSET, default=logging.INFO,
help="suppress output of execution log message.")
return group
[docs]def getGeneralOptionList(options):
option_list = []
try:
if options.dry_run:
option_list.append("--dry-run")
except AttributeError:
pass
if not options.with_no_log:
option_list.append("--logging")
if options.log_level == logging.DEBUG:
option_list.append("--debug")
if options.log_level == logging.NOTSET:
option_list.append("--quiet")
if options.stacktrace:
option_list.append("--stacktrace")
return option_list