Refactor OAuth provider linking and unlinking logic into a dedicated service; enhance error handling and logging throughout the application; improve sound management and scanning services with better file handling and unique naming; implement centralized error and logging services for consistent API responses and application-wide logging configuration.
This commit is contained in:
136
app/services/logging_service.py
Normal file
136
app/services/logging_service.py
Normal file
@@ -0,0 +1,136 @@
|
||||
"""Centralized logging service for the application."""
|
||||
|
||||
import logging
|
||||
import sys
|
||||
|
||||
|
||||
class LoggingService:
|
||||
"""Service for configuring and managing application logging."""
|
||||
|
||||
@staticmethod
|
||||
def setup_logging(
|
||||
level: str = "INFO",
|
||||
format_string: str | None = None,
|
||||
) -> None:
|
||||
"""Setup application-wide logging configuration."""
|
||||
if format_string is None:
|
||||
format_string = (
|
||||
"%(asctime)s - %(name)s - %(levelname)s - %(message)s"
|
||||
)
|
||||
|
||||
# Configure root logger
|
||||
logging.basicConfig(
|
||||
level=getattr(logging, level.upper()),
|
||||
format=format_string,
|
||||
handlers=[
|
||||
logging.StreamHandler(sys.stdout),
|
||||
],
|
||||
)
|
||||
|
||||
# Set specific logger levels for third-party libraries
|
||||
logging.getLogger("werkzeug").setLevel(logging.WARNING)
|
||||
logging.getLogger("urllib3").setLevel(logging.WARNING)
|
||||
logging.getLogger("requests").setLevel(logging.WARNING)
|
||||
|
||||
@staticmethod
|
||||
def get_logger(name: str) -> logging.Logger:
|
||||
"""Get a logger instance for a specific module."""
|
||||
return logging.getLogger(name)
|
||||
|
||||
@staticmethod
|
||||
def log_operation_start(logger: logging.Logger, operation: str) -> None:
|
||||
"""Log the start of an operation."""
|
||||
logger.info(f"Starting {operation}")
|
||||
|
||||
@staticmethod
|
||||
def log_operation_success(
|
||||
logger: logging.Logger,
|
||||
operation: str,
|
||||
details: str | None = None,
|
||||
) -> None:
|
||||
"""Log successful completion of an operation."""
|
||||
message = f"Successfully completed {operation}"
|
||||
if details:
|
||||
message += f" - {details}"
|
||||
logger.info(message)
|
||||
|
||||
@staticmethod
|
||||
def log_operation_error(
|
||||
logger: logging.Logger,
|
||||
operation: str,
|
||||
error: Exception,
|
||||
) -> None:
|
||||
"""Log an error during an operation."""
|
||||
logger.error(f"Error during {operation}: {error}")
|
||||
|
||||
@staticmethod
|
||||
def log_validation_error(
|
||||
logger: logging.Logger,
|
||||
field: str,
|
||||
value: str,
|
||||
reason: str,
|
||||
) -> None:
|
||||
"""Log validation errors consistently."""
|
||||
logger.warning(f"Validation failed for {field}='{value}': {reason}")
|
||||
|
||||
@staticmethod
|
||||
def log_resource_not_found(
|
||||
logger: logging.Logger,
|
||||
resource_type: str,
|
||||
identifier: str,
|
||||
) -> None:
|
||||
"""Log when a resource is not found."""
|
||||
logger.warning(f"{resource_type} not found: {identifier}")
|
||||
|
||||
@staticmethod
|
||||
def log_resource_created(
|
||||
logger: logging.Logger,
|
||||
resource_type: str,
|
||||
identifier: str,
|
||||
) -> None:
|
||||
"""Log when a resource is created."""
|
||||
logger.info(f"Created {resource_type}: {identifier}")
|
||||
|
||||
@staticmethod
|
||||
def log_resource_updated(
|
||||
logger: logging.Logger,
|
||||
resource_type: str,
|
||||
identifier: str,
|
||||
) -> None:
|
||||
"""Log when a resource is updated."""
|
||||
logger.info(f"Updated {resource_type}: {identifier}")
|
||||
|
||||
@staticmethod
|
||||
def log_resource_deleted(
|
||||
logger: logging.Logger,
|
||||
resource_type: str,
|
||||
identifier: str,
|
||||
) -> None:
|
||||
"""Log when a resource is deleted."""
|
||||
logger.info(f"Deleted {resource_type}: {identifier}")
|
||||
|
||||
@staticmethod
|
||||
def log_user_action(
|
||||
logger: logging.Logger,
|
||||
user_id: str,
|
||||
action: str,
|
||||
resource: str | None = None,
|
||||
) -> None:
|
||||
"""Log user actions for auditing."""
|
||||
message = f"User {user_id} performed action: {action}"
|
||||
if resource:
|
||||
message += f" on {resource}"
|
||||
logger.info(message)
|
||||
|
||||
@staticmethod
|
||||
def log_security_event(
|
||||
logger: logging.Logger,
|
||||
event_type: str,
|
||||
details: str,
|
||||
user_id: str | None = None,
|
||||
) -> None:
|
||||
"""Log security-related events."""
|
||||
message = f"Security event [{event_type}]: {details}"
|
||||
if user_id:
|
||||
message += f" (User: {user_id})"
|
||||
logger.warning(message)
|
||||
Reference in New Issue
Block a user