Bài 17: Logging - Ghi Log

Mục Tiêu Bài Học

Sau khi hoàn thành bài này, bạn sẽ:

✅ Sử dụng logging module
✅ Hiểu log levels
✅ Làm việc với handlers và formatters
✅ Configure logging
✅ Áp dụng best practices
✅ Sử dụng structured logging

Tại Sao Cần Logging?

Logging giúp:

Debug - Tìm bugs và hiểu program flow
Monitor - Track application health
Audit - Record user actions
Troubleshoot - Diagnose production issues

# ❌ Don't use print for productionprint("User logged in")  # Lost when app restarts # ✅ Use loggingimport logginglogging.info("User logged in")  # Persistent, configurable

1. Logging Basics

Log Levels

Python có 5 log levels (thấp đến cao):

import logging # DEBUG - Chi tiết thông tin cho debugginglogging.debug("Variable x = 10") # INFO - Thông tin chung về program flowlogging.info("Server started on port 8000") # WARNING - Cảnh báo, chưa phải lỗilogging.warning("Disk space low") # ERROR - Lỗi nghiêm trọnglogging.error("Failed to connect to database") # CRITICAL - Lỗi critical, app có thể crashlogging.critical("System out of memory")

Basic Configuration

import logging # Configure logginglogging.basicConfig(    level=logging.DEBUG,  # Show all logs >= DEBUG    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',    datefmt='%Y-%m-%d %H:%M:%S') # Now use logginglogging.debug("This is debug message")logging.info("This is info message")logging.warning("This is warning message")logging.error("This is error message")logging.critical("This is critical message") # Output:# 2024-01-01 10:00:00 - root - DEBUG - This is debug message# 2024-01-01 10:00:01 - root - INFO - This is info message# ...

Logging to File

import logging # Log to filelogging.basicConfig(    filename='app.log',    level=logging.INFO,    format='%(asctime)s - %(levelname)s - %(message)s') logging.info("Application started")logging.error("An error occurred") # Log to both console and filelogging.basicConfig(    level=logging.INFO,    format='%(asctime)s - %(levelname)s - %(message)s',    handlers=[        logging.FileHandler('app.log'),        logging.StreamHandler()  # Console    ])

2. Loggers, Handlers, Formatters

Loggers

import logging # Get logger với name (thường là __name__)logger = logging.getLogger(__name__)logger.setLevel(logging.DEBUG) # Module hierarchy# myapp.module1 -> logger = logging.getLogger(__name__)# myapp.module2 -> logger = logging.getLogger(__name__) # Sử dụng loggerlogger.debug("Debug message")logger.info("Info message")logger.warning("Warning message")logger.error("Error message")logger.critical("Critical message")

Handlers

Handlers xác định nơi logs được ghi.

import loggingfrom logging.handlers import RotatingFileHandler, TimedRotatingFileHandler logger = logging.getLogger(__name__)logger.setLevel(logging.DEBUG) # Console handlerconsole_handler = logging.StreamHandler()console_handler.setLevel(logging.INFO) # File handlerfile_handler = logging.FileHandler('app.log')file_handler.setLevel(logging.DEBUG) # Rotating file handler (rotate khi file đạt size)rotating_handler = RotatingFileHandler(    'app.log',    maxBytes=10*1024*1024,  # 10MB    backupCount=5  # Giữ 5 backup files) # Time-based rotating handler (rotate theo thời gian)time_handler = TimedRotatingFileHandler(    'app.log',    when='midnight',  # Rotate mỗi midnight    interval=1,    backupCount=7  # Giữ 7 days) # Add handlers to loggerlogger.addHandler(console_handler)logger.addHandler(file_handler)

Formatters

Formatters định dạng log messages.

import logging logger = logging.getLogger(__name__) # Create formatterformatter = logging.Formatter(    '%(asctime)s - %(name)s - %(levelname)s - %(message)s',    datefmt='%Y-%m-%d %H:%M:%S') # Create handler và set formatterhandler = logging.StreamHandler()handler.setFormatter(formatter) logger.addHandler(handler)logger.setLevel(logging.DEBUG) # Available format attributes:# %(name)s - Logger name# %(levelname)s - Log level (DEBUG, INFO, etc.)# %(message)s - Log message# %(asctime)s - Timestamp# %(filename)s - Source filename# %(funcName)s - Function name# %(lineno)d - Line number# %(pathname)s - Full pathname# %(process)d - Process ID# %(thread)d - Thread ID # Custom formatdetailed_formatter = logging.Formatter(    '[%(asctime)s] %(levelname)-8s [%(name)s:%(funcName)s:%(lineno)d] - %(message)s')

3. Advanced Configuration

Dictionary Configuration

import loggingimport logging.config LOGGING_CONFIG = {    'version': 1,    'disable_existing_loggers': False,        'formatters': {        'default': {            'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'        },        'detailed': {            'format': '[%(asctime)s] %(levelname)-8s [%(name)s:%(funcName)s:%(lineno)d] - %(message)s'        }    },        'handlers': {        'console': {            'class': 'logging.StreamHandler',            'level': 'INFO',            'formatter': 'default',            'stream': 'ext://sys.stdout'        },        'file': {            'class': 'logging.handlers.RotatingFileHandler',            'level': 'DEBUG',            'formatter': 'detailed',            'filename': 'app.log',            'maxBytes': 10485760,  # 10MB            'backupCount': 5        }    },        'loggers': {        '': {  # Root logger            'level': 'DEBUG',            'handlers': ['console', 'file']        },        'myapp': {            'level': 'DEBUG',            'handlers': ['console', 'file'],            'propagate': False        }    }} # Apply configurationlogging.config.dictConfig(LOGGING_CONFIG) # Use loggerlogger = logging.getLogger('myapp')logger.info("Application started")

File-based Configuration

logging.ini

[loggers]keys=root,myapp [handlers]keys=consoleHandler,fileHandler [formatters]keys=defaultFormatter,detailedFormatter [logger_root]level=DEBUGhandlers=consoleHandler [logger_myapp]level=DEBUGhandlers=consoleHandler,fileHandlerqualname=myapppropagate=0 [handler_consoleHandler]class=StreamHandlerlevel=INFOformatter=defaultFormatterargs=(sys.stdout,) [handler_fileHandler]class=handlers.RotatingFileHandlerlevel=DEBUGformatter=detailedFormatterargs=('app.log', 'a', 10485760, 5) [formatter_defaultFormatter]format=%(asctime)s - %(name)s - %(levelname)s - %(message)sdatefmt=%Y-%m-%d %H:%M:%S [formatter_detailedFormatter]format=[%(asctime)s] %(levelname)-8s [%(name)s:%(funcName)s:%(lineno)d] - %(message)s

import loggingimport logging.config # Load configuration từ filelogging.config.fileConfig('logging.ini') logger = logging.getLogger('myapp')logger.info("Logging configured from file")

4. Exception Logging

Basic Exception Logging

import logging logger = logging.getLogger(__name__) try:    result = 10 / 0except Exception as e:    # Log exception với traceback    logger.exception("An error occurred")        # Hoặc    logger.error("An error occurred", exc_info=True) # Output includes full traceback

Custom Exception Handler

import loggingimport sys logger = logging.getLogger(__name__) def handle_exception(exc_type, exc_value, exc_traceback):    """Handle uncaught exceptions"""    if issubclass(exc_type, KeyboardInterrupt):        # Don't log keyboard interrupt        sys.__excepthook__(exc_type, exc_value, exc_traceback)        return        logger.critical(        "Uncaught exception",        exc_info=(exc_type, exc_value, exc_traceback)    ) # Set exception hooksys.excepthook = handle_exception # Now all uncaught exceptions will be loggedraise ValueError("This will be logged!")

5. Contextual Logging

Logger Adapters

import logging class ContextAdapter(logging.LoggerAdapter):    """Add context to log messages"""        def process(self, msg, kwargs):        # Add extra context        return f"[{self.extra['user_id']}] {msg}", kwargs logger = logging.getLogger(__name__)logger.setLevel(logging.INFO)handler = logging.StreamHandler()logger.addHandler(handler) # Create adapter với contextcontext_logger = ContextAdapter(logger, {'user_id': '123'}) context_logger.info("User logged in")# Output: [123] User logged in

Contextual Variables

import logging logger = logging.getLogger(__name__) # Log với extra fieldslogger.info(    "User action",    extra={        'user_id': 123,        'action': 'login',        'ip_address': '192.168.1.1'    }) # Custom formatter để show extra fieldsclass CustomFormatter(logging.Formatter):    def format(self, record):        # Add custom fields        if hasattr(record, 'user_id'):            record.msg = f"[User {record.user_id}] {record.msg}"        return super().format(record) formatter = CustomFormatter('%(asctime)s - %(levelname)s - %(message)s')handler = logging.StreamHandler()handler.setFormatter(formatter)logger.addHandler(handler)

5 Ứng Dụng Thực Tế

1. Application Logger

import loggingfrom logging.handlers import RotatingFileHandler, SMTPHandlerfrom pathlib import Path class ApplicationLogger:    """Centralized application logger"""        def __init__(self, app_name: str, log_dir: str = "./logs"):        self.app_name = app_name        self.log_dir = Path(log_dir)        self.log_dir.mkdir(exist_ok=True)                self.logger = logging.getLogger(app_name)        self.logger.setLevel(logging.DEBUG)                self._setup_handlers()        def _setup_handlers(self):        """Setup logging handlers"""                # Console handler (INFO+)        console_handler = logging.StreamHandler()        console_handler.setLevel(logging.INFO)        console_formatter = logging.Formatter(            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'        )        console_handler.setFormatter(console_formatter)                # File handler for all logs (DEBUG+)        all_file_handler = RotatingFileHandler(            self.log_dir / 'app.log',            maxBytes=10*1024*1024,  # 10MB            backupCount=5        )        all_file_handler.setLevel(logging.DEBUG)        detailed_formatter = logging.Formatter(            '[%(asctime)s] %(levelname)-8s [%(name)s:%(funcName)s:%(lineno)d] - %(message)s'        )        all_file_handler.setFormatter(detailed_formatter)                # Error file handler (ERROR+)        error_file_handler = RotatingFileHandler(            self.log_dir / 'error.log',            maxBytes=10*1024*1024,            backupCount=5        )        error_file_handler.setLevel(logging.ERROR)        error_file_handler.setFormatter(detailed_formatter)                # Add handlers        self.logger.addHandler(console_handler)        self.logger.addHandler(all_file_handler)        self.logger.addHandler(error_file_handler)        def get_logger(self, module_name: str = None):        """Get logger for specific module"""        if module_name:            return logging.getLogger(f"{self.app_name}.{module_name}")        return self.logger        def add_email_handler(        self,        mailhost: tuple,        fromaddr: str,        toaddrs: list,        subject: str,        credentials: tuple = None    ):        """Add email handler for critical errors"""        email_handler = SMTPHandler(            mailhost=mailhost,            fromaddr=fromaddr,            toaddrs=toaddrs,            subject=subject,            credentials=credentials,            secure=()        )        email_handler.setLevel(logging.CRITICAL)        self.logger.addHandler(email_handler) # Sử dụngapp_logger = ApplicationLogger("myapp") # Get logger for different modulesauth_logger = app_logger.get_logger("auth")db_logger = app_logger.get_logger("database") auth_logger.info("User authentication started")db_logger.debug("Database query executed")auth_logger.error("Authentication failed") # Add email notifications cho critical errorsapp_logger.add_email_handler(    mailhost=('smtp.gmail.com', 587),    fromaddr='app@example.com',    toaddrs=['admin@example.com'],    subject='Critical Error in MyApp',    credentials=('username', 'password'))

2. Request Logger cho Web Apps

import loggingimport timefrom functools import wrapsfrom typing import Callable class RequestLogger:    """Log HTTP requests with timing"""        def __init__(self):        self.logger = logging.getLogger("request")        self.logger.setLevel(logging.INFO)                handler = logging.FileHandler("requests.log")        formatter = logging.Formatter(            '%(asctime)s - %(levelname)s - %(message)s'        )        handler.setFormatter(formatter)        self.logger.addHandler(handler)        def log_request(self, func: Callable) -> Callable:        """Decorator to log requests"""        @wraps(func)        def wrapper(*args, **kwargs):            # Extract request info (example for Flask)            request = kwargs.get('request') or args[0]                        start_time = time.time()                        # Log request            self.logger.info(                f"Request started",                extra={                    'method': getattr(request, 'method', 'UNKNOWN'),                    'path': getattr(request, 'path', '/'),                    'ip': getattr(request, 'remote_addr', 'unknown')                }            )                        try:                # Execute function                response = func(*args, **kwargs)                status_code = getattr(response, 'status_code', 200)                                # Log success                duration = time.time() - start_time                self.logger.info(                    f"Request completed",                    extra={                        'method': getattr(request, 'method', 'UNKNOWN'),                        'path': getattr(request, 'path', '/'),                        'status_code': status_code,                        'duration': f"{duration:.3f}s"                    }                )                                return response                            except Exception as e:                # Log error                duration = time.time() - start_time                self.logger.error(                    f"Request failed: {str(e)}",                    extra={                        'method': getattr(request, 'method', 'UNKNOWN'),                        'path': getattr(request, 'path', '/'),                        'duration': f"{duration:.3f}s"                    },                    exc_info=True                )                raise                return wrapper # Sử dụng với Flaskfrom flask import Flask, request app = Flask(__name__)request_logger = RequestLogger() @app.route('/api/users')@request_logger.log_requestdef get_users():    # Your logic here    return {'users': []}

3. Performance Logger

import loggingimport timefrom functools import wrapsfrom typing import Callable class PerformanceLogger:    """Log function performance metrics"""        def __init__(self):        self.logger = logging.getLogger("performance")        self.logger.setLevel(logging.INFO)                handler = logging.FileHandler("performance.log")        formatter = logging.Formatter(            '%(asctime)s - %(message)s'        )        handler.setFormatter(formatter)        self.logger.addHandler(handler)        def measure(self, threshold: float = 1.0) -> Callable:        """Decorator to measure function execution time"""        def decorator(func: Callable) -> Callable:            @wraps(func)            def wrapper(*args, **kwargs):                start_time = time.time()                                try:                    result = func(*args, **kwargs)                    duration = time.time() - start_time                                        # Log if exceeds threshold                    if duration >= threshold:                        self.logger.warning(                            f"{func.__name__} took {duration:.3f}s (threshold: {threshold}s)",                            extra={                                'function': func.__name__,                                'duration': duration,                                'threshold': threshold                            }                        )                    else:                        self.logger.info(                            f"{func.__name__} took {duration:.3f}s",                            extra={                                'function': func.__name__,                                'duration': duration                            }                        )                                        return result                                    except Exception as e:                    duration = time.time() - start_time                    self.logger.error(                        f"{func.__name__} failed after {duration:.3f}s: {str(e)}",                        exc_info=True                    )                    raise                        return wrapper        return decorator # Sử dụngperf_logger = PerformanceLogger() @perf_logger.measure(threshold=0.5)def slow_function():    time.sleep(1)    return "Done" @perf_logger.measure(threshold=2.0)def fast_function():    time.sleep(0.1)    return "Done" slow_function()  # Logs warning (> 0.5s)fast_function()  # Logs info (< 2.0s)

4. Database Query Logger

import loggingimport timefrom typing import Any, Optional class DatabaseLogger:    """Log database queries với performance tracking"""        def __init__(self):        self.logger = logging.getLogger("database")        self.logger.setLevel(logging.DEBUG)                # File handler        handler = logging.FileHandler("queries.log")        formatter = logging.Formatter(            '[%(asctime)s] %(levelname)s - %(message)s'        )        handler.setFormatter(formatter)        self.logger.addHandler(handler)                # Slow query threshold (seconds)        self.slow_query_threshold = 1.0        def log_query(        self,        query: str,        params: Optional[tuple] = None,        duration: Optional[float] = None    ):        """Log database query"""        log_data = {            'query': query,            'params': params        }                if duration:            log_data['duration'] = f"{duration:.3f}s"                        if duration >= self.slow_query_threshold:                self.logger.warning(                    f"Slow query detected ({duration:.3f}s): {query}",                    extra=log_data                )            else:                self.logger.debug(                    f"Query executed ({duration:.3f}s)",                    extra=log_data                )        else:            self.logger.debug(f"Query: {query}", extra=log_data)        def log_transaction(self, action: str, table: str, affected_rows: int = 0):        """Log database transaction"""        self.logger.info(            f"Transaction: {action} on {table}",            extra={                'action': action,                'table': table,                'affected_rows': affected_rows            }        )        def log_connection(self, event: str, details: str = ""):        """Log connection events"""        self.logger.info(            f"Connection {event}: {details}",            extra={'event': event, 'details': details}        ) # Sử dụngdb_logger = DatabaseLogger() # Wrapper cho database operationsclass DatabaseConnection:    def __init__(self, logger: DatabaseLogger):        self.logger = logger        def execute(self, query: str, params: tuple = ()):        """Execute query với logging"""        start_time = time.time()                try:            # Simulate query execution            # result = cursor.execute(query, params)            time.sleep(0.1)  # Simulate                        duration = time.time() - start_time            self.logger.log_query(query, params, duration)                        return "result"                    except Exception as e:            self.logger.logger.error(                f"Query failed: {query}",                extra={'query': query, 'params': params},                exc_info=True            )            raise db = DatabaseConnection(db_logger)db_logger.log_connection("established", "localhost:5432")db.execute("SELECT * FROM users WHERE id = %s", (123,))

5. Audit Logger

import loggingfrom datetime import datetimefrom typing import Dict, Any, Optional class AuditLogger:    """Log user actions cho audit trail"""        def __init__(self):        self.logger = logging.getLogger("audit")        self.logger.setLevel(logging.INFO)                # Separate audit log file        handler = logging.FileHandler("audit.log")        formatter = logging.Formatter(            '%(asctime)s - %(message)s'        )        handler.setFormatter(formatter)        self.logger.addHandler(handler)        def log_action(        self,        user_id: int,        action: str,        resource_type: str,        resource_id: Optional[int] = None,        details: Optional[Dict[str, Any]] = None,        ip_address: Optional[str] = None    ):        """Log user action"""        log_entry = {            'timestamp': datetime.utcnow().isoformat(),            'user_id': user_id,            'action': action,            'resource_type': resource_type,        }                if resource_id:            log_entry['resource_id'] = resource_id                if details:            log_entry['details'] = details                if ip_address:            log_entry['ip_address'] = ip_address                self.logger.info(            f"User {user_id} performed {action} on {resource_type}",            extra=log_entry        )        def log_login(self, user_id: int, success: bool, ip_address: str):        """Log login attempt"""        status = "successful" if success else "failed"        self.log_action(            user_id=user_id,            action=f"login_{status}",            resource_type="authentication",            ip_address=ip_address        )        def log_data_access(        self,        user_id: int,        resource_type: str,        resource_id: int,        action: str = "read"    ):        """Log data access"""        self.log_action(            user_id=user_id,            action=action,            resource_type=resource_type,            resource_id=resource_id        )        def log_data_modification(        self,        user_id: int,        resource_type: str,        resource_id: int,        action: str,        changes: Dict[str, Any]    ):        """Log data modification với changes"""        self.log_action(            user_id=user_id,            action=action,            resource_type=resource_type,            resource_id=resource_id,            details={'changes': changes}        ) # Sử dụngaudit_logger = AuditLogger() # Log loginaudit_logger.log_login(    user_id=123,    success=True,    ip_address="192.168.1.1") # Log data accessaudit_logger.log_data_access(    user_id=123,    resource_type="document",    resource_id=456,    action="read") # Log data modificationaudit_logger.log_data_modification(    user_id=123,    resource_type="user_profile",    resource_id=123,    action="update",    changes={        'email': {'old': 'old@example.com', 'new': 'new@example.com'},        'phone': {'old': None, 'new': '+1234567890'}    }) # Log deletionaudit_logger.log_action(    user_id=123,    action="delete",    resource_type="document",    resource_id=456,    details={'reason': 'User requested deletion'})

Best Practices

1. Logger Naming

# ✅ Use __name__ for module-based loggerslogger = logging.getLogger(__name__) # ✅ Hierarchical naming# myapp.auth.login -> myapp.auth -> myapp -> rootlogger = logging.getLogger('myapp.auth.login') # ❌ Don't use hardcoded stringslogger = logging.getLogger('my_logger')  # Not maintainable

2. Log Levels

# ✅ Use appropriate levelslogger.debug("Variable value: %s", variable)  # Developmentlogger.info("User logged in")  # Normal flowlogger.warning("API rate limit approaching")  # Potential issueslogger.error("Failed to save data")  # Errorslogger.critical("Database connection lost")  # Critical failures # ❌ Don't overuse levelslogger.error("User logged in")  # Wrong level!

3. String Formatting

# ✅ Use lazy formatting (efficient)logger.info("User %s logged in at %s", username, timestamp) # ❌ Don't format before logginglogger.info(f"User {username} logged in at {timestamp}")  # Always evaluated!

4. Sensitive Data

# ❌ Don't log sensitive datalogger.info(f"User password: {password}")  # Security issue!logger.info(f"Credit card: {cc_number}") # ✅ Mask sensitive datalogger.info(f"User registered with email: {email[:3]}***")logger.info(f"Card ending in: {cc_number[-4:]}")

Bài Tập Thực Hành

Bài 1: Basic Logger

Setup logging cho Python application với console và file handlers.

Bài 2: Rotating Logs

Implement rotating file handler với size-based và time-based rotation.

Bài 3: Request Logger

Tạo request logger cho Flask/FastAPI application.

Bài 4: Performance Monitoring

Implement decorator để log function execution time.

Bài 5: Audit Trail

Tạo audit logging system để track user actions.

Tóm Tắt

Trong bài này chúng ta đã học:

✅ Logging Basics - Log levels, basic configuration
✅ Components - Loggers, handlers, formatters
✅ Advanced Config - Dictionary config, file config
✅ Exception Logging - Traceback, uncaught exceptions
✅ Contextual Logging - Adapters, extra fields
✅ Real Applications - App logger, request logger, performance, database, audit

Logging là essential cho production applications - giúp debug, monitor, và troubleshoot!

Bài tiếp theo: Bài 18: Performance Optimization - Học cách tối ưu performance của Python applications! ⚡

#monitoring

Logging - Ghi Log