KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use rotating_log_handler method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

config.py

Source:config.py

1import sys2from dotenv import load_dotenv3import os4import logging5from logging.handlers import RotatingFileHandler6# load explicitly from the same directory as this file7load_dotenv(dotenv_path=os.path.join(os.path.dirname(__file__), ".env"))8SPREADSHEET_ID = os.getenv("SPREADSHEET_ID")9SCOPES = ["https://www.googleapis.com/auth/spreadsheets"]10SERVICE_ACCOUNT = os.path.join(os.path.dirname(__file__), "service_account.json")11EMAIL_RANGE = os.getenv("EMAIL_RANGE")12EMAILS_ROW_START = int(EMAIL_RANGE[1])13EMAIL_REGEX = r"^[^@+]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$"14CODE_RANGE = os.getenv("CODE_RANGE")15# subtract the value of the letter column from the value of the start column to get the numeric position of the column16CODES_COL_START = ord(CODE_RANGE[0]) - ord("A") + 117MAIL_SERVER = os.getenv("MAIL_SERVER")18MAIL_PORT = int(os.getenv("MAIL_PORT"))19MAIL_USERNAME = os.getenv("MAIL_USERNAME")20MAIL_PASSWORD = os.getenv("MAIL_PASSWORD")21MAIL_USE_TLS = os.getenv("MAIL_USE_TLS") == "True"22MAIL_USE_SSL = os.getenv("MAIL_USE_SSL") == "True"23MAIL_DEFAULT_SENDER_NAME = os.getenv("MAIL_DEFAULT_SENDER_NAME")24MAIL_DEFAULT_SENDER_EMAIL = os.getenv("MAIL_DEFAULT_SENDER_EMAIL")25# logging26log_formatter = logging.Formatter(27    "%(asctime)s %(levelname)s %(funcName)s(%(lineno)d) %(message)s"28)29log_file = os.path.join(os.path.dirname(__file__), "logs", "api.log")30if not os.path.exists(os.path.dirname(log_file)):31    os.makedirs(os.path.dirname(log_file))32rotating_log_handler = RotatingFileHandler(33    log_file,34    mode="a",35    maxBytes=5 * 1024 * 1024,36    backupCount=2,37    encoding=None,38    delay=0,39)40rotating_log_handler.setFormatter(log_formatter)41rotating_log_handler.setLevel(logging.INFO)42logger = logging.getLogger()43logger.setLevel(logging.INFO)...

misc.py

Source:misc.py

1import hashlib2import json3import logging4import os5import sys6from logging.handlers import TimedRotatingFileHandler7from settings import LOG_FILE, LOG_LEVEL8def setup_logger(name, base_level=LOG_LEVEL):9    logger = logging.getLogger(name)10    logger.setLevel(base_level)11    formatter = logging.Formatter(fmt='%(asctime)s %(levelname)-6s %(message)s',12                                  datefmt='%Y-%m-%d %H:%M:%S')13    rotating_log_handler = TimedRotatingFileHandler(14        LOG_FILE, when='d', backupCount=5)15    rotating_log_handler.setLevel(logging.DEBUG)16    rotating_log_handler.setFormatter(formatter)17    logger.addHandler(rotating_log_handler)18    stdout_log_handler = logging.StreamHandler(sys.stdout)19    stdout_log_handler.setLevel(logging.INFO)20    logger.addHandler(stdout_log_handler)21    return logger22def hash_file(file, algorithm='sha1', bufsize=8192):23    h = hashlib.new(algorithm)24    with open(file, 'rb') as f:25        block = f.read(bufsize)26        # TODO what if file is less than 8192 kbytes?27        # if not block:28        #     break29        h.update(block)30    return h.hexdigest()31def get_pretty_json(json_data):32    return json.dumps(json_data, indent=2)33def url_to_filename(url):...

logger.py

Source:logger.py

1import logging2from enum import Enum3from logging.handlers import RotatingFileHandler4from airflow.configuration import conf5BASE_LOGGING_FORMAT = (6    "[%(asctime)s] %(levelname)-8s {%(name)s:%(module)s:%(lineno)d} - %(message)s"7)8class Loggers(Enum):9    API_ROUTES = "AirflowAPIRoutes"10    API = "AirflowAPI"11    OPERATIONS = "AirflowOperations"12    WORKFLOW = "AirflowWorkflow"13def build_logger(logger_name: str) -> logging.Logger:14    logger = logging.getLogger(logger_name)15    log_format = logging.Formatter(BASE_LOGGING_FORMAT)16    rotating_log_handler = RotatingFileHandler(17        f"{conf.get('logging', 'base_log_folder', fallback='')}/openmetadata_airflow_api.log",18        maxBytes=1000000,19        backupCount=10,20    )21    rotating_log_handler.setFormatter(log_format)22    logger.addHandler(rotating_log_handler)23    logger.setLevel(logging.DEBUG)24    return logger25def routes_logger() -> logging.Logger:26    return build_logger(Loggers.API_ROUTES.value)27def api_logger():28    return build_logger(Loggers.API.value)29def operations_logger():30    return build_logger(Loggers.OPERATIONS.value)31def workflow_logger():...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.