def configure_standard_logging(verbosity: int, mode: LoggingMode):
"""Configure the standard library's `logging` module.
Get `logging` working with options consistent with Twisted. NOTE CAREFULLY
that `django.utils.log.DEFAULT_LOGGING` may have been applied (though only
if installed and if configured in this environment). Those settings and
the settings this function applies must be mentally combined to understand
the resultant behaviour.
:param verbosity: See `get_logging_level`.
:param mode: The mode in which to configure logging. See `LoggingMode`.
"""
set_standard_verbosity(verbosity)
# Make sure that `logging` is not configured to capture warnings.
logging.captureWarnings(False)
# If a logger is ever configured `propagate=False` but without handlers
# `logging.Logger.callHandlers` will employ the `lastResort` handler in
# order that the log is not lost. This goes to standard error by default.
# Here we arrange for these situations to be logged more distinctively so
# that they're easier to diagnose.
logging.lastResort = (
logging.StreamHandler(
twistedModern.LoggingFile(
logger=twistedModern.Logger("lost+found"),
level=twistedModern.LogLevel.error)))
python类captureWarnings()的实例源码
def predict(
predictor: str,
verbose: bool,
):
# Logging setup.
logging.captureWarnings(True)
if verbose:
logging.root.setLevel(logging.DEBUG)
# Read labels.
labels = read_binary()
labels = acton.proto.wrappers.LabelPool.deserialise(labels)
# Write predictions.
proto = acton.acton.predict(labels=labels, predictor=predictor)
write_binary(proto.proto.SerializeToString())
# acton-recommend
def recommend(
diversity: float,
recommendation_count: int,
recommender: str,
verbose: bool,
):
# Logging setup.
logging.warning('Not implemented: diversity')
logging.captureWarnings(True)
if verbose:
logging.root.setLevel(logging.DEBUG)
# Read the predictions protobuf.
predictions = read_binary()
predictions = acton.proto.wrappers.Predictions.deserialise(predictions)
# Write the recommendations protobuf.
proto = acton.acton.recommend(
predictions=predictions,
recommender=recommender,
n_recommendations=recommendation_count)
write_binary(proto.proto.SerializeToString())
# acton-label
def test_benchmark():
"""Benchmark reading/writing a file and output the time"""
logging.captureWarnings(True)
FINPUT = os.path.join(PWD, 'data', 'NSL_catalog_col.txt')
FOUTPUT = "/tmp/eqclusterng-test-output.txt"
import time
t0 = time.time()
t = BPTree.from_file(FINPUT) # Init tree with events from a file
t.grow() # Populate B-P tree with events
t.prune() # Prune given cutoff (calculate n if none)
t.output2file(FOUTPUT) # Output to file to match MATLAB perf
t1 = time.time()
print " bench: {0}eq/{1:.6f}s ".format(len(t.P), t1-t0),
def configure_logging(logging_config, logging_settings):
if not sys.warnoptions:
# Route warnings through python logging
logging.captureWarnings(True)
# RemovedInNextVersionWarning is a subclass of DeprecationWarning which
# is hidden by default, hence we force the "default" behavior
warnings.simplefilter("default", RemovedInNextVersionWarning)
if logging_config:
# First find the logging configuration function ...
logging_config_func = import_string(logging_config)
dictConfig(DEFAULT_LOGGING)
# ... then invoke it with the logging settings
if logging_settings:
logging_config_func(logging_settings)
def tearDown(self):
# Be sure to reset the warning capture
logging.captureWarnings(False)
super(TestDaiquiri, self).tearDown()
def setup(level=logging.WARNING, outputs=[output.STDERR], program_name=None,
capture_warnings=True):
"""Setup Python logging.
This will setup basic handlers for Python logging.
:param level: Root log level.
:param outputs: Iterable of outputs to log to.
:param program_name: The name of the program. Auto-detected if not set.
:param capture_warnings: Capture warnings from the `warnings' module.
"""
root_logger = logging.getLogger(None)
# Remove all handlers
for handler in list(root_logger.handlers):
root_logger.removeHandler(handler)
# Add configured handlers
for out in outputs:
if isinstance(out, str):
out = output.preconfigured.get(out)
if out is None:
raise RuntimeError("Output {} is not available".format(out))
out.add_to_logger(root_logger)
root_logger.setLevel(level)
program_logger = logging.getLogger(program_name)
def logging_excepthook(exc_type, value, tb):
program_logger.critical(
"".join(traceback.format_exception(exc_type, value, tb)))
sys.excepthook = logging_excepthook
if capture_warnings:
logging.captureWarnings(True)
def capture_logging():
# mostly warnings caused by self-signed certs
logging.captureWarnings(True)
def main():
"""Entry point"""
logging.config.fileConfig(pkg_resources.resource_filename(
'functest', 'ci/logging.ini'))
logging.captureWarnings(True)
LOGGER.info('Starting Functest server')
api_add_resource()
init_db()
APP.run(host='0.0.0.0')
def main():
"""Entry point"""
logging.config.fileConfig(pkg_resources.resource_filename(
'functest', 'ci/logging.ini'))
logging.captureWarnings(True)
deployment = CheckDeployment()
return deployment.check_all()
def main():
"""Entry point"""
logging.config.fileConfig(pkg_resources.resource_filename(
'functest', 'ci/logging.ini'))
logging.captureWarnings(True)
parser = RunTestsParser()
args = parser.parse_args(sys.argv[1:])
runner = Runner()
return runner.main(**args).value
def main():
"""
main entry point for script
"""
opts = getoptions()
logging.basicConfig(format='%(asctime)s [%(levelname)s] %(message)s', datefmt='%Y-%m-%dT%H:%M:%S', level=opts['log'])
if sys.version.startswith("2.7"):
logging.captureWarnings(True)
config = Config(opts['config'])
dwconfig = config.getsection("datawarehouse")
dbif = DbHelper(dwconfig, 'modw_supremm.batchscripts')
for resourcename, settings in config.resourceconfigs():
if opts['resource'] in (None, resourcename, str(settings['resource_id'])):
logging.debug("Processing %s (id=%s)", resourcename, settings['resource_id'])
if "script_dir" in settings:
total = processfor(settings['resource_id'], settings['script_dir'], dbif, opts['deltadays'])
logging.info("Processed %s files for %s", total, resourcename)
else:
logging.debug("Skip resource %s no script dir defined", resourcename)
dbif.postinsert()
def main():
"""
main entry point for script
"""
opts, args = getoptions()
logging.basicConfig(format='%(asctime)s [%(levelname)s] %(message)s', datefmt='%Y-%m-%dT%H:%M:%S', level=opts['log'])
if sys.version.startswith("2.7"):
logging.captureWarnings(True)
preprocs = loadpreprocessors()
plugins = loadplugins()
if len(opts['plugin_whitelist']) > 0:
preprocs, plugins = filter_plugins({"plugin_whitelist": opts['plugin_whitelist']}, preprocs, plugins)
elif len(opts['plugin_blacklist']) > 0:
preprocs, plugins = filter_plugins({"plugin_blacklist": opts['plugin_blacklist']}, preprocs, plugins)
logging.debug("Loaded %s preprocessors", len(preprocs))
logging.debug("Loaded %s plugins", len(plugins))
archivelist = args
job = MockJob(archivelist)
config = Config(confpath=opts['config'])
preprocessors = [x(job) for x in preprocs]
analytics = [x(job) for x in plugins]
s = Summarize(preprocessors, analytics, job, config)
s.process()
result = s.get()
print json.dumps(result, indent=4)
def __init__(self, capture_warnings=True, use_default_kvp=True, json=False):
self._config = deepcopy(DEFAULT_LOGGING_CONF)
if use_default_kvp:
self.update_default_formatter(DEFAULT_KVP_FORMAT)
if json:
self.enable_json_formatter()
# Custom level to suppress handlers
logging.addLevelName('DISABLED', LEVEL_MAP['DISABLED'])
logging.captureWarnings(capture_warnings)
def setup_logging(conf):
# Add additional dependent libraries that have unhelp bug levels
extra_log_level_defaults = []
logging.set_defaults(default_log_levels=logging.get_default_log_levels() +
extra_log_level_defaults)
logging.setup(conf, 'deckhand')
py_logging.captureWarnings(True)
def verbosity(lvl):
if lvl < 1:
return
elif lvl == 1:
log = logging.getLogger('aptrepo')
log.setLevel(logging.INFO)
elif lvl == 2:
log = logging.getLogger()
log.setLevel(logging.INFO)
else:
log = logging.getLogger()
log.setLevel(logging.DEBUG)
log.addHandler(logging.StreamHandler())
logging.captureWarnings(True)
def set_logger(log_file_path):
logger = logging.getLogger(NAME)
logger.setLevel(logging.INFO)
file_handler = logging.handlers.RotatingFileHandler(log_file_path, maxBytes=1024 * 1024, backupCount=10)
file_handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)-8s %(message)s"))
logger.addHandler(file_handler)
logging.captureWarnings(True)
return logger
def init_logger(self) -> None:
try:
log_path = QStandardPaths.writableLocation(QStandardPaths.AppConfigLocation)
except AttributeError:
if sys.platform == 'win32':
log_path = os.path.join(QDir.homePath(), 'AppData', 'Local', qApp.applicationName().lower())
elif sys.platform == 'darwin':
log_path = os.path.join(QDir.homePath(), 'Library', 'Preferences', qApp.applicationName().lower())
else:
log_path = os.path.join(QDir.homePath(), '.config', qApp.applicationName().lower())
os.makedirs(log_path, exist_ok=True)
self.console = ConsoleWidget(self)
self.consoleLogger = ConsoleHandler(self.console)
handlers = [logging.handlers.RotatingFileHandler(os.path.join(log_path, '%s.log'
% qApp.applicationName().lower()),
maxBytes=1000000, backupCount=1),
self.consoleLogger]
if self.parser.isSet(self.debug_option) or self.verboseLogs:
# noinspection PyTypeChecker
handlers.append(logging.StreamHandler())
logging.setLoggerClass(VideoLogger)
logging.basicConfig(handlers=handlers,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M',
level=logging.INFO)
logging.captureWarnings(capture=True)
sys.excepthook = MainWindow.log_uncaught_exceptions
def tweet_string(message, log, media=None):
check_twitter_config()
logging.captureWarnings(True)
old_level = log.getEffectiveLevel()
log.setLevel(logging.ERROR)
twitter = Twython(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
retries = 0
while retries < 5:
log.setLevel(logging.ERROR)
try:
if media:
photo = open(media, 'rb')
media_ids = twitter.upload_media(media=photo)
twitter.update_status(status=message.encode('utf-8').strip(), media_ids=media_ids['media_id'])
else:
twitter.update_status(status=message.encode('utf-8').strip())
break
except TwythonAuthError, e:
log.setLevel(old_level)
log.exception(" Problem trying to tweet string")
twitter_auth_issue(e)
return
except:
log.setLevel(old_level)
log.exception(" Problem trying to tweet string")
retries += 1
s = random.randrange(5, 10 * retries)
log.debug(" sleeping %d seconds for retry", s)
time.sleep(s)
log.setLevel(old_level)
if retries == 5:
log.error("Couldn't tweet string: %s with media: %s", message, media)
def get_following(log, id):
log.debug(" Getting people %s is following", id)
check_twitter_config()
logging.captureWarnings(True)
old_level = log.getEffectiveLevel()
log.setLevel(logging.ERROR)
twitter = Twython(APP_KEY, APP_SECRET, OAUTH_TOKEN, OAUTH_TOKEN_SECRET)
log.setLevel(old_level)
cursor = -1
max_loops = 15
while cursor != 0:
try:
log.setLevel(logging.ERROR)
following = twitter.get_friends_list(screen_name=id, cursor=cursor, count=200)
log.setLevel(old_level)
except TwythonAuthError, e:
log.exception(" Problem trying to get people following")
twitter_auth_issue(e)
raise
except:
raise
for u in following["users"]:
yield u["screen_name"]
cursor = following["next_cursor"]
if cursor:
s = random.randint(55, 65)
log.debug(" Sleeping %ds to avoid rate limit. Cursor: %s", s, cursor)
time.sleep(s)
else:
log.debug(" Normal query end")
max_loops -= 1
if max_loops <= 0:
log.debug(" Killing search due to max loops")
break
log.setLevel(old_level)
def execute(self, *args, **options):
if options['verbosity'] > 0:
# ensure that deprecation warnings are displayed during testing
# the following state is assumed:
# logging.capturewarnings is true
# a "default" level warnings filter has been added for
# DeprecationWarning. See django.conf.LazySettings._configure_logging
logger = logging.getLogger('py.warnings')
handler = logging.StreamHandler()
logger.addHandler(handler)
super(Command, self).execute(*args, **options)
if options['verbosity'] > 0:
# remove the testing-specific handler
logger.removeHandler(handler)
def run(self):
if self.verbosity > 0:
# ensure that deprecation warnings are displayed during testing
# the following state is assumed:
# logging.capturewarnings is true
# a "default" level warnings filter has been added for
# DeprecationWarning. See django.conf.LazySettings._configure_logging
logger = logging.getLogger('py.warnings')
handler = logging.StreamHandler()
logger.addHandler(handler)
TestCommand.run(self)
if self.verbosity > 0:
# remove the testing-specific handler
logger.removeHandler(handler)
def execute(self, *args, **options):
if options['verbosity'] > 0:
# ensure that deprecation warnings are displayed during testing
# the following state is assumed:
# logging.capturewarnings is true
# a "default" level warnings filter has been added for
# DeprecationWarning. See django.conf.LazySettings._configure_logging
logger = logging.getLogger('py.warnings')
handler = logging.StreamHandler()
logger.addHandler(handler)
super(Command, self).execute(*args, **options)
if options['verbosity'] > 0:
# remove the testing-specific handler
logger.removeHandler(handler)
def execute(self, *args, **options):
if options['verbosity'] > 0:
# ensure that deprecation warnings are displayed during testing
# the following state is assumed:
# logging.capturewarnings is true
# a "default" level warnings filter has been added for
# DeprecationWarning. See django.conf.LazySettings._configure_logging
logger = logging.getLogger('py.warnings')
handler = logging.StreamHandler()
logger.addHandler(handler)
super(Command, self).execute(*args, **options)
if options['verbosity'] > 0:
# remove the testing-specific handler
logger.removeHandler(handler)
def execute(self, *args, **options):
if options['verbosity'] > 0:
# ensure that deprecation warnings are displayed during testing
# the following state is assumed:
# logging.capturewarnings is true
# a "default" level warnings filter has been added for
# DeprecationWarning. See django.conf.LazySettings._configure_logging
logger = logging.getLogger('py.warnings')
handler = logging.StreamHandler()
logger.addHandler(handler)
super(Command, self).execute(*args, **options)
if options['verbosity'] > 0:
# remove the testing-specific handler
logger.removeHandler(handler)
def execute(self, *args, **options):
if options['verbosity'] > 0:
# ensure that deprecation warnings are displayed during testing
# the following state is assumed:
# logging.capturewarnings is true
# a "default" level warnings filter has been added for
# DeprecationWarning. See django.conf.LazySettings._configure_logging
logger = logging.getLogger('py.warnings')
handler = logging.StreamHandler()
logger.addHandler(handler)
super(Command, self).execute(*args, **options)
if options['verbosity'] > 0:
# remove the testing-specific handler
logger.removeHandler(handler)
def execute(self, *args, **options):
if options['verbosity'] > 0:
# ensure that deprecation warnings are displayed during testing
# the following state is assumed:
# logging.capturewarnings is true
# a "default" level warnings filter has been added for
# DeprecationWarning. See django.conf.LazySettings._configure_logging
logger = logging.getLogger('py.warnings')
handler = logging.StreamHandler()
logger.addHandler(handler)
super(Command, self).execute(*args, **options)
if options['verbosity'] > 0:
# remove the testing-specific handler
logger.removeHandler(handler)
def execute(self, *args, **options):
if options['verbosity'] > 0:
# ensure that deprecation warnings are displayed during testing
# the following state is assumed:
# logging.capturewarnings is true
# a "default" level warnings filter has been added for
# DeprecationWarning. See django.conf.LazySettings._configure_logging
logger = logging.getLogger('py.warnings')
handler = logging.StreamHandler()
logger.addHandler(handler)
super(Command, self).execute(*args, **options)
if options['verbosity'] > 0:
# remove the testing-specific handler
logger.removeHandler(handler)
def execute(self, *args, **options):
if options['verbosity'] > 0:
# ensure that deprecation warnings are displayed during testing
# the following state is assumed:
# logging.capturewarnings is true
# a "default" level warnings filter has been added for
# DeprecationWarning. See django.conf.LazySettings._configure_logging
logger = logging.getLogger('py.warnings')
handler = logging.StreamHandler()
logger.addHandler(handler)
super(Command, self).execute(*args, **options)
if options['verbosity'] > 0:
# remove the testing-specific handler
logger.removeHandler(handler)
def execute(self, *args, **options):
if options['verbosity'] > 0:
# ensure that deprecation warnings are displayed during testing
# the following state is assumed:
# logging.capturewarnings is true
# a "default" level warnings filter has been added for
# DeprecationWarning. See django.conf.LazySettings._configure_logging
logger = logging.getLogger('py.warnings')
handler = logging.StreamHandler()
logger.addHandler(handler)
super(Command, self).execute(*args, **options)
if options['verbosity'] > 0:
# remove the testing-specific handler
logger.removeHandler(handler)
