aws-logging-handlers

import logging import os from aws_logging_handlers.S3 import S3Handler from aws_logging_handlers.Kinesis import KinesisHandler # Configure AWS credentials (replace with your actual bucket and stream names) bucket_name = os.environ.get('AWS_S3_LOG_BUCKET', 'your-s3-log-bucket') kinesis_stream_name = os.environ.get('AWS_KINESIS_LOG_STREAM', 'your-kinesis-log-stream') aws_region = os.environ.get('AWS_REGION', 'us-east-1') # Configure S3 handler (logs rotate every 5MB or 120 seconds) s3_handler = S3Handler( log_group='test_log_s3', bucket_name=bucket_name, workers=3, # Number of upload worker threads session_kwargs={'region_name': aws_region} ) # Configure Kinesis handler kinesis_handler = KinesisHandler( log_group='test_log_kinesis', stream_name=kinesis_stream_name, workers=1, # Number of upload worker threads session_kwargs={'region_name': aws_region} ) # Set up formatter formatter = logging.Formatter( '[%(asctime)s] %(filename)s:%(lineno)d} %(levelname)s - %(message)s' ) s3_handler.setFormatter(formatter) kinesis_handler.setFormatter(formatter) # Get logger and add handlers logger = logging.getLogger('my_app') logger.setLevel(logging.INFO) logger.addHandler(s3_handler) logger.addHandler(kinesis_handler) # Log some messages logger.info("This is an info message to S3 and Kinesis.") logger.warning("A warning occurred in the application.") logger.error("An error message that should go to AWS services.") # Ensure all buffered logs are flushed and workers shut down gracefully logging.shutdown()