janus-logging
sync and async logging within one logger instance.
Installation
pip install janus-loggingUsage
Default
import asyncio
import logging
import sys
import janus_logging
def threaded(sync_log, counter: int):
for i in range(counter):
sync_log.info('s-Hello #%s' % i)
sync_log.info('s-Finished #%s' % counter)
async def async_coro(async_log, counter: int):
for i in range(counter):
await async_log.info('aio-Hello #%s' % i)
await async_log.info('aio-Finished #%s' % counter)
#
counter = 4
name = 'my_janus_logger'
level = logging.DEBUG
stream = sys.stdout
loop = asyncio.get_event_loop()
#
logger = janus_logging.JanusLogger(name=name, level=level, loop=loop, stream=stream)
loop.run_until_complete(
asyncio.gather(
loop.run_in_executor(
None,
threaded,
logger.logger_sync(),
counter
),
async_coro(
logger.logger_async(),
counter
)
)
)
logger.shutdown()
#
#
loop.close()The output of above will look like:
s-Hello #0
s-Hello #1
s-Hello #2
aio-Hello #0
s-Hello #3
s-Finished #4
aio-Hello #1
aio-Hello #2
aio-Hello #3
aio-Finished #4Please note, that the output might be different on your instance.
JSON
Simply use fixtures.
import asyncio
import logging
import sys
import janus_logging
def threaded(sync_log, counter: int):
for i in range(counter):
sync_log.info(
's-Hello #%s' % i,
extra=dict(counter=i, log_type='sync', log_status='in progress')
)
sync_log.info(
's-Finished #%s' % counter,
extra=dict(total=counter, log_type='sync', log_status='finished')
)
async def async_coro(async_log, counter: int):
for i in range(counter):
await async_log.info(
'aio-Hello #%s' % i,
extra=dict(counter=i, log_type='async', log_status='in progress')
)
await async_log.info(
'aio-Finished #%s' % counter,
extra=dict(total=counter, log_type='async', log_status='finished')
)
#
counter = 4
name = 'my_janus_logger'
level = logging.DEBUG
stream = sys.stdout
loop = asyncio.get_event_loop()
#
logger = janus_logging.JanusLogger(
name=name,
level=level,
loop=loop,
fixture=janus_logging.fixture_json,
stream=stream,
extra=dict(bla='blabla')
)
loop.run_until_complete(
asyncio.gather(
loop.run_in_executor(
None,
threaded,
logger.logger_sync(logger_name='logger_sync'),
counter
),
async_coro(
logger.logger_async(logger_name='logger_async'),
counter
)
)
)
logger.shutdown()
#
#
loop.close()The output of above will look like:
{"level": "INFO", "msg": "s-Hello #0", "log_type": "sync", "bla": "blabla", "logger_name": "logger_sync", "counter": 0, "log_status": "in progress", "logged_at": "2019-09-27T12:00:02.517101+02:00", "line_numer": 35, "function": "threaded", "file_path": "demo_janus_log.py"}
{"level": "INFO", "msg": "aio-Hello #1", "log_type": "async", "logged_at": "2019-09-27T12:00:02.518000+02:00", "line_number": 60, "function": "info", "file_path": "/home/madkote/janus-logging/janus_logging/__init__.py", "bla": "blabla", "logger_name": "logger_async", "counter": 1, "log_status": "in progress"}
...Custom
If a custom logger, formatter, handler are required, then create custom
fixtures and pass them to the JanusLogger.
def fixture_custom(
name: str,
level: int,
loop: loop: asyncio.AbstractEventLoop,
**kwargs
) -> logging.Logger:
...
return ...
logger = janus_logging.JanusLogger(
...,
fixture=fixture_custom,
...
)Development
Issues and suggestions are welcome through issues
