enables Queues if multiprocessing.Queue() is set in the "log_queue" setting
Now a logger "Log.init_worker_logging" can be attached to the ProcessPoolExecutor,
the init args must be the log_queue set
example:
```py
with concurrent.futures.ProcessPoolExecutor(
max_workers=max_forks,
initializer=Log.init_worker_logging,
initargs=(log_queue,)
) as executor:
````
Move all settings into a settings argument, the structure is defined in LogSettings.
Default settings are in Log.DEFAULT_LOG_SETTINGS
Only log path and log name are parameters
Color output for console is on default enabled, disable via "console_color_output_enabled"
The complete console output can be stopped with "console_enabled"
49 lines
1.3 KiB
Python
49 lines
1.3 KiB
Python
"""
|
|
Log logging_handling.log testing
|
|
"""
|
|
|
|
# import atexit
|
|
from pathlib import Path
|
|
from multiprocessing import Queue
|
|
# this is for testing only
|
|
from queue_logger.log_queue import QueueLogger
|
|
from corelibs.logging_handling.log import Log
|
|
|
|
|
|
def main():
|
|
"""
|
|
Log testing
|
|
"""
|
|
script_path: Path = Path(__file__).resolve().parent
|
|
log = Log(
|
|
log_path=script_path.joinpath('log', 'test.log'),
|
|
log_name="Test Log",
|
|
log_settings={
|
|
"log_level_console": 'DEBUG',
|
|
"log_level_file": 'DEBUG',
|
|
# "console_color_output_enabled": False,
|
|
}
|
|
)
|
|
|
|
log.logger.debug('Debug test: %s', log.logger.name)
|
|
log.logger.info('Info test: %s', log.logger.name)
|
|
log.logger.warning('Warning test: %s', log.logger.name)
|
|
log.logger.error('Error test: %s', log.logger.name)
|
|
log.logger.critical('Critical test: %s', log.logger.name)
|
|
log.exception('Exception test: %s', log.logger.name)
|
|
|
|
log_queue: 'Queue[str]' = Queue()
|
|
log_q = QueueLogger(
|
|
log_file=script_path.joinpath('log', 'test_queue.log'),
|
|
log_name="Test Log Queue",
|
|
log_queue=log_queue
|
|
)
|
|
log_q.mlog.info('Log test: %s', log.logger.name)
|
|
# log_q.stop_listener()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|
|
# __END__
|