PKmJ"M̎requests_handler/__init__.py""" Handler for logging records to a Restful API. :param api_url: URL to post records to. :param auth: Credentials in format (username, password). :param level: Logging level (DEBUG, INFO, ERROR, WARNING, CRITICAL). :param filter: Minimum record level to emit. :param bubble: Should bubble. """ import json import arrow import requests import records from logbook import LogRecord, Handler __version__ = "0.0.8" class RequestsHandler(Handler): """ Handler for logging records to a Restful API. :param logger: str to use for the logger name. :param api_url: URL to post records to. :param auth: Credentials in format (username, password). :param level: Logging level (DEBUG, INFO, ERROR, WARNING, CRITICAL). :param filter: Minimum record level to emit. :param bubble: Should bubble. """ def __init__(self, api_url: str, auth: tuple, level: str = 'INFO', logger: str = None, filter: str = None, bubble: bool = False): Handler.__init__(self, level, filter, bubble) self.api_url = api_url self.auth = auth self.logger = logger self.log_record_fields = [ "logger", "channel", "level", "lineno", "filename", "module", "func_name", "message", "log_time" ] def emit(self, record: LogRecord): log_record = record.to_dict() log_record["log_time"] = arrow.utcnow().strftime( "%Y-%m-%d %H:%M:%S.%f") log_record['logger'] = self.logger log_record = {k: v for k, v in log_record.items() if k in self.log_record_fields} try: requests.post(url=self.api_url, data=log_record, auth=self.auth) except Exception as e: print(e) class DatabaseHandler(Handler): """ Handler for logging records to a database. :param connection_string: database connection string. :param level: Logging level (DEBUG, INFO, ERROR, WARNING, CRITICAL). :param filter: Minimum record level to emit. :param bubble: Should bubble. """ def __init__(self, connection_string: str, process_id: int, level: str = 'INFO', filter: str = None, bubble: bool = False): Handler.__init__(self, level, filter, bubble) self.connection_string = connection_string self.process_id = process_id self.log_record_fields = [ "channel", "level", "lineno", "filename", "module", "func_name", "message", "log_time" ] self.db = records.Database(self.connection_string) def emit(self, record: LogRecord): log_record = record.to_dict() log_record["log_time"] = arrow.utcnow().strftime( "%Y-%m-%d %H:%M:%S.%f") log_record = {k: v for k, v in log_record.items() if k in self.log_record_fields} log_record['message'] = log_record.get("message").replace("'", '"') try: self.db.query("INSERT INTO data_processors.logs" "(process_id, channel, level, lineno, filename, " "module, func_name, message, log_time)" f"VALUES({self.process_id}, " f"'{log_record.get('channel')}'," f"{log_record.get('level')}," f"{log_record.get('lineno')}," f"'{log_record.get('filename')}'," f"'{log_record.get('module')}'," f"'{log_record.get('func_name')}'," f"'{log_record.get('message')}'," f"'{log_record.get('log_time')}')") except Exception as e: print(e) def main(): pass if __name__ == '__main__': main() PK}JG!requests_handler/test_handlers.pyfrom time import sleep import logbook from iologger import iologger from requests_handler import RequestsHandler, DatabaseHandler def test_iologger_database(): @iologger() def my_func(string: str, integer: int) -> str: sleep(1.25) return f"my string was {string} and my integer was {integer}." database_handler = DatabaseHandler(f"postgresql://data_loads_user:HILLS!@localhost:5433/data_loads", process_id=12, level='DEBUG') database_handler.push_application() my_func("Abraham", integer=15) my_func("Isaac", 0) database_handler.pop_application() def test_handler(): logger = logbook.Logger("TestLogger") requests_handler = RequestsHandler( "https://autosweet-data.com/api/v1/logs", auth=('admin', 'aut0sw33t')) requests_handler.push_application() logger.debug("A debug message") logger.critical("A critical message!") requests_handler.pop_application() def test_iologger(): @iologger() def my_func(string: str, integer: int) -> str: return f"my string was {string} and my integer was {integer}." requests_handler = RequestsHandler( "https://autosweet-data.com/api/v1/logs", auth=('admin', 'aut0sw33t'), logger='TestLogger' ) requests_handler.push_application() my_func("Abraham", integer=15) my_func("Isaac", 0) requests_handler.pop_application() def main(): pass if __name__ == '__main__': test_iologger_database() PK!H|&Ub&requests_handler-0.0.8.dist-info/WHEEL HM K-*ϳR03rOK-J,/RH,Q034 /, (-JLR()*M ILR(4KM̫#DPK!HwJp)requests_handler-0.0.8.dist-info/METADATAuAs0:4x =4CBC#[LG`q޷oW[f #ӆc_$+ffAW&]'AJcIN}kԍ! {gs|%KCg2br+5E}LfG5덫{}=`;%y+R& /L/?M\dJʒd?߷۟g9wpLzō- hGngR 9Pj,`}Eލ m ?YudeFg(dN=QLQKg}p St"M^!-RP)xYudR.(h19.2w2:קb"qz`f#KP ^M!^PK!HJE'requests_handler-0.0.8.dist-info/RECORDMo0ỿ0`ᰃ|ёUj4U2](Bly+Q^Ejv2?ꃱB1҂-輘[wMVWnQY|q>iWvr+5Z/~Lx9q9]1u?{ s-l;:fyQ+EQ4WxD}.1lk縑tc <1dQ|%q,Mb'~ t0& 7 PKmJ"M̎requests_handler/__init__.pyPK}JG!requests_handler/test_handlers.pyPK!H|&Ub&&requests_handler-0.0.8.dist-info/WHEELPK!HwJp)requests_handler-0.0.8.dist-info/METADATAPK!HJE'requests_handler-0.0.8.dist-info/RECORDPK