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='[email protected]', toaddrs=['[email protected]'], 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': '[email protected]', 'new': '[email protected]'}, '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! ⚡