Django REST Framework - Modul 5
Von Basic Logging bis Production Monitoring
"Es funktioniert nicht!" → Aber warum? 🤷♂️
Mit Logging:
[2024-01-15 14:23:45] ERROR - Movie creation failed: Duplicate title "Inception"
[2024-01-15 14:23:45] INFO - User: john_doe (ID: 42)
[2024-01-15 14:23:45] DEBUG - Request Data: {"title": "Inception", "year": 2010}
import logging
# 5 Standard Log Levels:
logging.DEBUG # 10 - Detaillierte Infos (nur Development)
logging.INFO # 20 - Allgemeine Infos
logging.WARNING # 30 - Warnung (Standard)
logging.ERROR # 40 - Fehler (funktioniert noch)
logging.CRITICAL # 50 - Kritischer Fehler (funktioniert nicht mehr)
import logging
# Logger erstellen
logger = logging.getLogger(__name__)
# Logging
logger.debug("Detailed information for debugging")
logger.info("General information")
logger.warning("Something unexpected happened")
logger.error("An error occurred")
logger.critical("Critical error - system unstable")
def create_movie(data):
print("Creating movie:", data)
# ...
print("Movie created successfully!")
# Nachteile:
# - Keine Log Levels
# - Schwer zu filtern
# - Keine Timestamps
# - Nicht konfigurierbar
import logging
logger = logging.getLogger(__name__)
def create_movie(data):
logger.info("Creating movie: %s", data)
# ...
logger.info("Movie created successfully!")
# Vorteile:
# - Log Levels
# - Filterbar
# - Timestamps
# - Konfigurierbar
# filepath: movieapi/settings.py
import os
LOGGING = {
'version': 1,
'disable_existing_loggers': False,
# Formatters: Wie sehen Log-Nachrichten aus?
'formatters': {
'verbose': {
'format': '{levelname} {asctime} {module} {process:d} {thread:d} {message}',
'style': '{',
},
'simple': {
'format': '{levelname} {message}',
'style': '{',
},
},
# Handlers: Wohin gehen die Logs?
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'formatter': 'simple',
},
'file': {
'class': 'logging.FileHandler',
'filename': os.path.join(BASE_DIR, 'logs', 'django.log'),
'formatter': 'verbose',
},
},
# Loggers: Welche Logs werden erfasst?
'loggers': {
'django': {
'handlers': ['console', 'file'],
'level': 'INFO',
'propagate': True,
},
},
}
# Terminal:
mkdir logs
# .gitignore:
logs/
*.log
# filepath: movieapi/settings.py
import os
LOGGING = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'verbose': {
'format': '[{levelname}] {asctime} {module} {funcName} - {message}',
'style': '{',
'datefmt': '%Y-%m-%d %H:%M:%S',
},
'simple': {
'format': '[{levelname}] {message}',
'style': '{',
},
'json': {
'class': 'pythonjsonlogger.jsonlogger.JsonFormatter',
'format': '%(asctime)s %(name)s %(levelname)s %(message)s',
},
},
'filters': {
'require_debug_true': {
'class': 'django.utils.log.RequireDebugTrue',
},
'require_debug_false': {
'class': 'django.utils.log.RequireDebugFalse',
},
},
'handlers': {
# Console Handler (Development)
'console': {
'level': 'DEBUG',
'class': 'logging.StreamHandler',
'formatter': 'simple',
'filters': ['require_debug_true'],
},
# File Handler (All Logs)
'file': {
'level': 'INFO',
'class': 'logging.handlers.RotatingFileHandler',
'filename': os.path.join(BASE_DIR, 'logs', 'django.log'),
'maxBytes': 1024 * 1024 * 10, # 10 MB
'backupCount': 5,
'formatter': 'verbose',
},
# Error File Handler
'error_file': {
'level': 'ERROR',
'class': 'logging.handlers.RotatingFileHandler',
'filename': os.path.join(BASE_DIR, 'logs', 'errors.log'),
'maxBytes': 1024 * 1024 * 10, # 10 MB
'backupCount': 5,
'formatter': 'verbose',
},
# API Handler (nur für API Logs)
'api_file': {
'level': 'INFO',
'class': 'logging.handlers.RotatingFileHandler',
'filename': os.path.join(BASE_DIR, 'logs', 'api.log'),
'maxBytes': 1024 * 1024 * 10, # 10 MB
'backupCount': 5,
'formatter': 'verbose',
},
# Security Handler
'security_file': {
'level': 'WARNING',
'class': 'logging.handlers.RotatingFileHandler',
'filename': os.path.join(BASE_DIR, 'logs', 'security.log'),
'maxBytes': 1024 * 1024 * 10, # 10 MB
'backupCount': 5,
'formatter': 'verbose',
},
# Mail Handler (Production - bei Errors Email senden)
'mail_admins': {
'level': 'ERROR',
'class': 'django.utils.log.AdminEmailHandler',
'filters': ['require_debug_false'],
},
},
'loggers': {
# Django Default Logger
'django': {
'handlers': ['console', 'file'],
'level': 'INFO',
'propagate': False,
},
# Django Request Logger
'django.request': {
'handlers': ['error_file', 'mail_admins'],
'level': 'ERROR',
'propagate': False,
},
# Django Security Logger
'django.security': {
'handlers': ['security_file'],
'level': 'WARNING',
'propagate': False,
},
# DRF Logger
'rest_framework': {
'handlers': ['api_file', 'console'],
'level': 'INFO',
'propagate': False,
},
# Movies App Logger
'movies': {
'handlers': ['api_file', 'console', 'error_file'],
'level': 'DEBUG',
'propagate': False,
},
},
# Root Logger (Fallback)
'root': {
'handlers': ['console', 'file'],
'level': 'INFO',
},
}
# filepath: movies/views.py
import logging
from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticatedOrReadOnly
from django.core.exceptions import ValidationError
from .models import Movie
from .serializers import MovieSerializer
# Logger für movies App
logger = logging.getLogger('movies')
class MovieViewSet(viewsets.ModelViewSet):
"""ViewSet für Movie CRUD mit Logging"""
queryset = Movie.objects.all()
serializer_class = MovieSerializer
permission_classes = [IsAuthenticatedOrReadOnly]
def list(self, request, *args, **kwargs):
"""Liste aller Filme - mit Logging"""
logger.info("Movie list requested by user: %s", request.user)
try:
response = super().list(request, *args, **kwargs)
logger.info("Movie list returned %d items", len(response.data['results']))
return response
except Exception as e:
logger.error("Error retrieving movie list: %s", str(e), exc_info=True)
raise
def retrieve(self, request, *args, **kwargs):
"""Einzelner Film - mit Logging"""
movie_id = kwargs.get('pk')
logger.debug("Movie detail requested: ID=%s, User=%s", movie_id, request.user)
try:
response = super().retrieve(request, *args, **kwargs)
logger.info("Movie retrieved: %s", response.data.get('title'))
return response
except Exception as e:
logger.error("Error retrieving movie ID=%s: %s", movie_id, str(e))
raise
def create(self, request, *args, **kwargs):
"""Film erstellen - mit Logging"""
logger.info("Movie creation attempt by user: %s", request.user)
logger.debug("Movie data: %s", request.data)
try:
response = super().create(request, *args, **kwargs)
logger.info(
"Movie created successfully: %s (ID=%s) by %s",
response.data.get('title'),
response.data.get('id'),
request.user
)
return response
except ValidationError as e:
logger.warning(
"Movie creation validation failed: %s - Data: %s",
str(e),
request.data
)
raise
except Exception as e:
logger.error(
"Movie creation failed: %s - User: %s - Data: %s",
str(e),
request.user,
request.data,
exc_info=True
)
raise
def update(self, request, *args, **kwargs):
"""Film aktualisieren - mit Logging"""
movie_id = kwargs.get('pk')
logger.info("Movie update attempt: ID=%s by %s", movie_id, request.user)
logger.debug("Update data: %s", request.data)
try:
response = super().update(request, *args, **kwargs)
logger.info(
"Movie updated successfully: %s (ID=%s)",
response.data.get('title'),
movie_id
)
return response
except Exception as e:
logger.error(
"Movie update failed: ID=%s, Error: %s",
movie_id,
str(e),
exc_info=True
)
raise
def destroy(self, request, *args, **kwargs):
"""Film löschen - mit Logging"""
movie_id = kwargs.get('pk')
try:
movie = self.get_object()
movie_title = movie.title
logger.warning(
"Movie deletion attempt: %s (ID=%s) by %s",
movie_title,
movie_id,
request.user
)
response = super().destroy(request, *args, **kwargs)
logger.warning(
"Movie deleted: %s (ID=%s) by %s",
movie_title,
movie_id,
request.user
)
return response
except Exception as e:
logger.error(
"Movie deletion failed: ID=%s, Error: %s",
movie_id,
str(e),
exc_info=True
)
raise
@action(detail=False, methods=['get'])
def top_rated(self, request):
"""Top-rated Movies - mit Logging"""
logger.info("Top-rated movies requested by %s", request.user)
try:
movies = Movie.objects.filter(rating__gte=8.0).order_by('-rating')[:10]
serializer = self.get_serializer(movies, many=True)
logger.info("Top-rated movies returned: %d items", len(serializer.data))
return Response(serializer.data)
except Exception as e:
logger.error("Error retrieving top-rated movies: %s", str(e), exc_info=True)
return Response(
{'error': 'Failed to retrieve top-rated movies'},
status=status.HTTP_500_INTERNAL_SERVER_ERROR
)
# filepath: movies/middleware.py
import logging
import time
import json
from django.utils.deprecation import MiddlewareMixin
logger = logging.getLogger('movies')
class RequestLoggingMiddleware(MiddlewareMixin):
"""
Middleware zum Loggen aller API Requests
"""
def process_request(self, request):
"""Request loggen"""
# Start-Zeit speichern
request._start_time = time.time()
# Request Info loggen
logger.info(
"API Request: %s %s from %s",
request.method,
request.path,
self._get_client_ip(request)
)
# Request Body loggen (nur bei POST/PUT/PATCH)
if request.method in ['POST', 'PUT', 'PATCH']:
try:
if request.content_type == 'application/json':
body = json.loads(request.body.decode('utf-8'))
# Passwörter maskieren
if 'password' in body:
body['password'] = '***HIDDEN***'
logger.debug("Request Body: %s", body)
except Exception as e:
logger.debug("Could not parse request body: %s", str(e))
return None
def process_response(self, request, response):
"""Response loggen"""
# Response-Zeit berechnen
if hasattr(request, '_start_time'):
duration = time.time() - request._start_time
else:
duration = 0
# Response loggen
logger.info(
"API Response: %s %s - Status: %d - Duration: %.2fs",
request.method,
request.path,
response.status_code,
duration
)
# Bei Errors: Response Body loggen
if response.status_code >= 400:
try:
if hasattr(response, 'data'):
logger.error("Error Response: %s", response.data)
except Exception:
pass
return response
def process_exception(self, request, exception):
"""Exceptions loggen"""
logger.error(
"Exception in API Request: %s %s - Error: %s",
request.method,
request.path,
str(exception),
exc_info=True
)
return None
def _get_client_ip(self, request):
"""Client IP ermitteln"""
x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
if x_forwarded_for:
ip = x_forwarded_for.split(',')[0]
else:
ip = request.META.get('REMOTE_ADDR')
return ip
# filepath: movieapi/settings.py
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
'django.middleware.common.CommonMiddleware',
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
'django.middleware.clickjacking.XFrameOptionsMiddleware',
# Custom Middleware
'movies.middleware.RequestLoggingMiddleware', # ← Hinzufügen!
]
# filepath: movies/views.py
import logging
from rest_framework.decorators import api_view, permission_classes
from rest_framework.response import Response
from rest_framework.authtoken.models import Token
from rest_framework.permissions import AllowAny, IsAuthenticated
from django.contrib.auth import authenticate
from django.contrib.auth.models import User
# Security Logger
security_logger = logging.getLogger('django.security')
api_logger = logging.getLogger('movies')
@api_view(['POST'])
@permission_classes([AllowAny])
def register(request):
"""User registrieren - mit Security Logging"""
username = request.data.get('username')
email = request.data.get('email')
api_logger.info("Registration attempt: username=%s, email=%s", username, email)
try:
# User erstellen
serializer = UserSerializer(data=request.data)
if serializer.is_valid():
user = serializer.save()
token, created = Token.objects.get_or_create(user=user)
# SUCCESS Log
api_logger.info(
"User registered successfully: %s (ID=%s)",
user.username,
user.id
)
security_logger.info(
"New user registration: %s from IP %s",
user.username,
_get_client_ip(request)
)
return Response({
'token': token.key,
'user': UserDetailSerializer(user).data
}, status=status.HTTP_201_CREATED)
# VALIDATION ERROR Log
api_logger.warning(
"Registration validation failed: username=%s, errors=%s",
username,
serializer.errors
)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
except Exception as e:
# ERROR Log
api_logger.error(
"Registration error: username=%s, error=%s",
username,
str(e),
exc_info=True
)
return Response(
{'error': 'Registration failed'},
status=status.HTTP_500_INTERNAL_SERVER_ERROR
)
@api_view(['POST'])
@permission_classes([AllowAny])
def login(request):
"""User login - mit Security Logging"""
username = request.data.get('username')
password = request.data.get('password')
ip_address = _get_client_ip(request)
api_logger.info("Login attempt: username=%s from %s", username, ip_address)
if not username or not password:
security_logger.warning(
"Login attempt with missing credentials from %s",
ip_address
)
return Response(
{'error': 'Please provide both username and password'},
status=status.HTTP_400_BAD_REQUEST
)
# Authentifizieren
user = authenticate(username=username, password=password)
if not user:
# FAILED LOGIN Log (Security!)
security_logger.warning(
"Failed login attempt: username=%s from %s",
username,
ip_address
)
api_logger.warning("Login failed: invalid credentials for %s", username)
return Response(
{'error': 'Invalid credentials'},
status=status.HTTP_401_UNAUTHORIZED
)
# Token erstellen/holen
token, created = Token.objects.get_or_create(user=user)
# SUCCESS Log
api_logger.info(
"User logged in successfully: %s (ID=%s) from %s",
user.username,
user.id,
ip_address
)
security_logger.info(
"Successful login: %s from %s",
user.username,
ip_address
)
return Response({
'token': token.key,
'user': UserDetailSerializer(user).data
})
@api_view(['POST'])
@permission_classes([IsAuthenticated])
def logout(request):
"""User logout - mit Security Logging"""
username = request.user.username
ip_address = _get_client_ip(request)
api_logger.info("Logout: user=%s from %s", username, ip_address)
try:
# Token löschen
request.user.auth_token.delete()
# SUCCESS Log
api_logger.info("User logged out successfully: %s", username)
security_logger.info("User logout: %s from %s", username, ip_address)
return Response({'message': 'Successfully logged out'})
except Exception as e:
# ERROR Log
api_logger.error(
"Logout error: user=%s, error=%s",
username,
str(e),
exc_info=True
)
return Response(
{'error': 'Logout failed'},
status=status.HTTP_500_INTERNAL_SERVER_ERROR
)
def _get_client_ip(request):
"""Helper: Client IP ermitteln"""
x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
if x_forwarded_for:
ip = x_forwarded_for.split(',')[0]
else:
ip = request.META.get('REMOTE_ADDR')
return ip
[INFO] API Request: GET /api/movies/ from 127.0.0.1
[INFO] Movie list requested by user: john_doe
[INFO] Movie list returned 10 items
[INFO] API Response: GET /api/movies/ - Status: 200 - Duration: 0.12s
[INFO] API Request: POST /api/movies/ from 127.0.0.1
[DEBUG] Request Body: {'title': 'Inception', 'year': 2010, 'genre': 'Sci-Fi'}
[INFO] Movie creation attempt by user: john_doe
[DEBUG] Movie data: {'title': 'Inception', 'year': 2010, 'genre': 'Sci-Fi'}
[INFO] Movie created successfully: Inception (ID=42) by john_doe
[INFO] API Response: POST /api/movies/ - Status: 201 - Duration: 0.08s
[WARNING] Login attempt with missing credentials from 192.168.1.100
[INFO] API Response: POST /api/auth/login/ - Status: 400 - Duration: 0.02s
[WARNING] Failed login attempt: username=hacker from 192.168.1.100
[WARNING] Login failed: invalid credentials for hacker
[INFO] API Response: POST /api/auth/login/ - Status: 401 - Duration: 0.15s
[INFO] 2024-01-15 14:23:45 views list - Movie list requested by user: john_doe
[INFO] 2024-01-15 14:23:45 views list - Movie list returned 10 items
[INFO] 2024-01-15 14:23:46 views create - Movie creation attempt by user: john_doe
[DEBUG] 2024-01-15 14:23:46 views create - Movie data: {'title': 'Inception', 'year': 2010}
[INFO] 2024-01-15 14:23:46 views create - Movie created successfully: Inception (ID=42) by john_doe
[WARNING] 2024-01-15 14:25:10 views login - Failed login attempt: username=hacker from 192.168.1.100
[WARNING] 2024-01-15 14:25:15 views login - Failed login attempt: username=admin from 192.168.1.100
[WARNING] 2024-01-15 14:25:20 views login - Failed login attempt: username=root from 192.168.1.100
[INFO] 2024-01-15 14:26:30 views login - Successful login: john_doe from 192.168.1.50
[INFO] 2024-01-15 14:30:00 views logout - User logout: john_doe from 192.168.1.50
[ERROR] 2024-01-15 14:27:00 views create - Movie creation failed: UNIQUE constraint failed: movies_movie.title - User: john_doe - Data: {'title': 'Inception', 'year': 2010}
Traceback (most recent call last):
File "movies/views.py", line 45, in create
response = super().create(request, *args, **kwargs)
...
django.db.utils.IntegrityError: UNIQUE constraint failed: movies_movie.title
# filepath: movies/decorators.py
import logging
import time
from functools import wraps
logger = logging.getLogger('movies')
def log_performance(func):
"""
Decorator zum Messen der Ausführungszeit
"""
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.time()
# Funktion ausführen
result = func(*args, **kwargs)
# Zeit berechnen
duration = time.time() - start_time
# Loggen
logger.info(
"Performance: %s.%s executed in %.2fs",
func.__module__,
func.__name__,
duration
)
# Bei langsamen Queries warnen
if duration > 1.0:
logger.warning(
"Slow function detected: %s.%s took %.2fs",
func.__module__,
func.__name__,
duration
)
return result
return wrapper
# filepath: movies/views.py
from .decorators import log_performance
class MovieViewSet(viewsets.ModelViewSet):
# ...
@log_performance
@action(detail=False, methods=['get'])
def complex_query(self, request):
"""Komplexe Query mit Performance Logging"""
# Komplexe Datenbank-Abfrage
movies = Movie.objects.select_related('owner').prefetch_related(
'castings__artist'
).filter(
rating__gte=8.0
).order_by('-rating')[:100]
serializer = self.get_serializer(movies, many=True)
return Response(serializer.data)
# filepath: movieapi/settings.py
# Development: DB Queries loggen
if DEBUG:
LOGGING['loggers']['django.db.backends'] = {
'handlers': ['console'],
'level': 'DEBUG',
'propagate': False,
}
# Output:
# (0.001) SELECT * FROM movies_movie WHERE id = 1;
# (0.023) SELECT * FROM movies_artist WHERE id IN (1, 2, 3, 4, 5);
# DEBUG: Nur Development
logger.debug("Detailed variable values")
# INFO: Production - Normal
logger.info("User created movie")
# WARNING: Attention needed
logger.warning("Failed login attempt")
# ERROR: Something went wrong
logger.error("Database error", exc_info=True)
# CRITICAL: System down
logger.critical("Database connection lost")
# ❌ NIEMALS loggen:
# - Passwörter
# - Tokens
# - Kreditkarten-Nummern
# - Persönliche Daten (DSGVO!)
# ✅ Maskieren:
def sanitize_log_data(data):
sensitive_fields = ['password', 'token', 'credit_card']
for field in sensitive_fields:
if field in data:
data[field] = '***HIDDEN***'
return data
logger.info("Request: %s", sanitize_log_data(request.data))
# Installation:
pip install python-json-logger
# settings.py:
'formatters': {
'json': {
'class': 'pythonjsonlogger.jsonlogger.JsonFormatter',
},
}
# Output (JSON):
{
"timestamp": "2024-01-15T14:23:45",
"level": "INFO",
"logger": "movies",
"message": "Movie created",
"movie_id": 42,
"user": "john_doe"
}
# RotatingFileHandler:
'handlers': {
'file': {
'class': 'logging.handlers.RotatingFileHandler',
'filename': 'logs/django.log',
'maxBytes': 1024 * 1024 * 10, # 10 MB
'backupCount': 5, # 5 Backup-Dateien
},
}
# Erstellt automatisch:
# django.log
# django.log.1
# django.log.2
# django.log.3
# django.log.4
# django.log.5
# Sentry (Error Tracking)
pip install sentry-sdk
# settings.py:
import sentry_sdk
from sentry_sdk.integrations.django import DjangoIntegration
sentry_sdk.init(
dsn="your-sentry-dsn",
integrations=[DjangoIntegration()],
environment="production",
)
# Logstash, Elasticsearch, Kibana (ELK Stack)
# Datadog, New Relic, CloudWatch
# settings.py (Production):
LOGGING = {
# ...
'loggers': {
'django': {
'level': 'WARNING', # Nur Warnings+
},
'movies': {
'level': 'INFO', # App Logs
},
'django.security': {
'level': 'WARNING', # Security!
},
},
}
# Development:
# level = 'DEBUG' # Alles loggen
logs/
├── django.log # Alle Django Logs
├── api.log # API-spezifische Logs
├── errors.log # Nur Errors
└── security.log # Security Events (Login, etc.)
Keep coding, keep learning! 💻