Skip to main content

Log Queue

Project description

logqueue

Log Queue

Initialize

Declare thread function.

def logger_function(log_dict):
    print(log_dict)
    # ...

output:
{'log_type': 'exception',
'timestamp': 1700000000.100001,
'process_id': 1234,
'thread_id': 1234567890,
'cpu_usage': 12, # if exist psutil
'memory_usage': 12, # if exist psutil
'file_name': 'test.py',
'file_lineno': 1,
'text': 'start',
'trace_back': 'error'} # if exception

import logqueue
logqueue.initialize(logger_function)
# ...

Close and Join

logqueue.close()
logqueue.join()

Signal

import signal
import logqueue

def signal_handler(_, frame):
    logqueue.close()

signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGABRT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)

# ... 
logqueue.join()

Logging

logqueue.put_info("start")
# or
logqueue.info("start")
log_str = logqueue.parse(log_dict)
print(log_str)

output:
2023-11-15 07:13:20.100001 12%:CPU 12%:Mem 234:PID 4567890:TID info test.py:1 start

**kwargs

logqueue.info("hi", alarm_meesage="alarm", input_database=True)
log_dict = logqueue.get()
print(log_dict)

output:
{'timestamp': 1700000000.100001,
'process_id': 1234,
'thread_id': 1234567890,
'log_type': 'information',
'file_name': 'test.py',
'file_lineno': 1,
'text': 'start',
'alarm_meesage': "alarm",
'input_database': True}

Log function types

logqueue.put(log_type:str, *objs:object, **kwargs)
logqueue.info(*objs:object, **kwargs) # or put_info()
logqueue.debug(*objs:object, **kwargs) # or put_debug()
logqueue.exception(*objs:object, **kwargs) # or put_exception()
logqueue.signal(*objs:object, **kwargs) # or put_signal()

Parse

get_log_formatters() # default log formatters
# {date} {time} {process_id:0{process_id_max_length}d}:PID {thread_id:0{thread_id_max_length}d}:TID {file_name:>{file_name_length}}:{file_lineno:<{file_lineno_length}} {log_type:{log_type_max_length}} {text}
clear_log_formatter()

append_log_formatter(f"{{{LogKey.date}}}")
append_log_formatter(f"{{{LogKey.time}}}")
get_log_formatters()
# {date} {time}
set_date_formatter("%y-%m-%d")
get_date_formatter()
# %y-%m-%d
set_process_id_formatter(f"{{{LogKey.process_id}:0{{{LogKey.process_id_max_length}}}d}}:PID")
get_process_id_formatter()
# {process_id:0{process_id_max_length}d}:PID

LogKeys

LogKey.date  
LogKey.time  
LogKey.timestamp  
LogKey.process_id  
LogKey.process_id_max_length  
LogKey.thread_id  
LogKey.thread_id_max_length  
LogKey.cpu_usage  
LogKey.memory_usage  
LogKey.log_type  
LogKey.log_type_max_length  
LogKey.file_info  
LogKey.file_name  
LogKey.file_name_length  
LogKey.file_lineno  
LogKey.file_lineno_length  
LogKey.text  
LogKey.trace_back  

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

logqueue-0.0.2.tar.gz (6.8 kB view hashes)

Uploaded Source

Built Distribution

logqueue-0.0.2-py3-none-any.whl (5.9 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page