logger = logging.getLogger(__name__)
logger_dump = logger.getChild('info')
# logger_dump = logging.getLogger(f'{__name__}.info') # сделало бы то же самоеclass MyHandler(logging.Handler):
def __init__(self, level=logging.NOTSET, ...): # в конструктор передаёшь подключение к БД, имя таблицы ну и что там ещё надо
super().__init__(level)
... # тут сохраняешь полученное подключение, проверяешь наличие таблицы и т.п.
def emit(self, record: logging.LogRecord) -> None:
# этот метод будет вызван уже после применения фильтров: по уровню и кастомных
... # пишешь тут логику записи в БД. Логами внутри этого метода пользоваться нельзя!
class MyFilter(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
# тут определяешь, представляет ли для тебя интерес сообщение record
return True # если представляет, верни True, иначе Falseloghandler = MyHandler(logging.NOTSET, ...) # экземпляр твоего логописца
logfilter = MyFilter() # экземпляр фильтра
loghandler.addFilter(logfilter) # привязываем фильтр
log = logger.getLogger(...)
log.addHandler(loghandler) # привязываем логописецПолагаете, что есть какое-то сопоставление вида "пользователь домашнего интернета подключён к IP 6.7.8.9 и оттуда же одновременно происходит коннект к такому-то сайту", и на основе этого принимается решение, что это ВПН?
import asyncio
import module_A # модуль A нужен воркеру
def worker_func(x): # функция-воркер
...
async def main():
import module_B
import module_C
import concurrent.futures as cf
...
if __name__ == '__main__':
asyncio.run(main())