200 lines
5.9 KiB
Python
200 lines
5.9 KiB
Python
"""
|
|
API-Utilities für konsistente Response-Strukturen
|
|
===============================================
|
|
|
|
Dieses Modul enthält Hilfsfunktionen zur Standardisierung von API-Responses
|
|
und zur Behandlung verschiedener Response-Formate im MYP-System.
|
|
"""
|
|
|
|
from datetime import datetime
|
|
from flask import jsonify
|
|
from typing import Dict, List, Any, Optional, Union
|
|
|
|
|
|
def create_success_response(data: Any = None, message: str = None, **kwargs) -> Dict[str, Any]:
|
|
"""
|
|
Erstellt eine standardisierte Erfolgs-Response.
|
|
|
|
Args:
|
|
data: Die zurückzugebenden Daten
|
|
message: Optionale Erfolgs-Nachricht
|
|
**kwargs: Zusätzliche Felder für die Response
|
|
|
|
Returns:
|
|
dict: Standardisierte Erfolgs-Response
|
|
"""
|
|
response = {
|
|
"success": True,
|
|
"timestamp": datetime.now().isoformat()
|
|
}
|
|
|
|
if data is not None:
|
|
response["data"] = data
|
|
|
|
if message:
|
|
response["message"] = message
|
|
|
|
# Zusätzliche Felder hinzufügen
|
|
response.update(kwargs)
|
|
|
|
return response
|
|
|
|
|
|
def create_error_response(error: str, details: str = None, **kwargs) -> Dict[str, Any]:
|
|
"""
|
|
Erstellt eine standardisierte Fehler-Response.
|
|
|
|
Args:
|
|
error: Haupt-Fehlermeldung
|
|
details: Detaillierte Fehlerinformationen
|
|
**kwargs: Zusätzliche Felder für die Response
|
|
|
|
Returns:
|
|
dict: Standardisierte Fehler-Response
|
|
"""
|
|
response = {
|
|
"success": False,
|
|
"error": error,
|
|
"timestamp": datetime.now().isoformat()
|
|
}
|
|
|
|
if details:
|
|
response["details"] = details
|
|
|
|
# Zusätzliche Felder hinzufügen
|
|
response.update(kwargs)
|
|
|
|
return response
|
|
|
|
|
|
def create_printers_response(printers: List[Dict], message: str = None) -> Dict[str, Any]:
|
|
"""
|
|
Erstellt eine standardisierte Response für Drucker-Listen.
|
|
|
|
Args:
|
|
printers: Liste der Drucker-Daten
|
|
message: Optionale Nachricht
|
|
|
|
Returns:
|
|
dict: Standardisierte Drucker-Response
|
|
"""
|
|
return create_success_response(
|
|
printers=printers,
|
|
count=len(printers),
|
|
message=message or "Drucker erfolgreich geladen"
|
|
)
|
|
|
|
|
|
def validate_printer_data(printer_dict: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Validiert und standardisiert Drucker-Daten.
|
|
|
|
Args:
|
|
printer_dict: Rohe Drucker-Daten
|
|
|
|
Returns:
|
|
dict: Validierte und standardisierte Drucker-Daten
|
|
"""
|
|
return {
|
|
"id": printer_dict.get("id"),
|
|
"name": printer_dict.get("name", "Unbekannter Drucker"),
|
|
"model": printer_dict.get("model") or "Unbekanntes Modell",
|
|
"location": printer_dict.get("location") or "Unbekannter Standort",
|
|
"status": printer_dict.get("status") or "offline",
|
|
"ip_address": printer_dict.get("ip_address"),
|
|
"plug_ip": printer_dict.get("plug_ip"),
|
|
"active": printer_dict.get("active", True),
|
|
"created_at": printer_dict.get("created_at"),
|
|
"last_checked": printer_dict.get("last_checked")
|
|
}
|
|
|
|
|
|
def handle_api_exception(error: Exception, context: str = "API-Operation") -> tuple:
|
|
"""
|
|
Behandelt API-Exceptions und erstellt konsistente Fehler-Responses.
|
|
|
|
Args:
|
|
error: Die aufgetretene Exception
|
|
context: Kontext der Operation für bessere Fehlermeldungen
|
|
|
|
Returns:
|
|
tuple: (response_dict, status_code)
|
|
"""
|
|
error_message = f"Fehler bei {context}"
|
|
|
|
response = create_error_response(
|
|
error=error_message,
|
|
details=str(error)
|
|
)
|
|
|
|
# Standard HTTP-Status für verschiedene Exception-Typen
|
|
status_code = 500
|
|
|
|
if "not found" in str(error).lower():
|
|
status_code = 404
|
|
elif "permission" in str(error).lower() or "unauthorized" in str(error).lower():
|
|
status_code = 403
|
|
elif "validation" in str(error).lower() or "invalid" in str(error).lower():
|
|
status_code = 400
|
|
|
|
return response, status_code
|
|
|
|
|
|
class ResponseValidator:
|
|
"""
|
|
Klasse zur Validierung und Standardisierung von API-Responses.
|
|
"""
|
|
|
|
@staticmethod
|
|
def is_valid_response(response_data: Dict[str, Any]) -> bool:
|
|
"""
|
|
Prüft, ob eine Response-Struktur gültig ist.
|
|
|
|
Args:
|
|
response_data: Zu prüfende Response-Daten
|
|
|
|
Returns:
|
|
bool: True wenn gültig
|
|
"""
|
|
if not isinstance(response_data, dict):
|
|
return False
|
|
|
|
# Minimal erforderliche Felder
|
|
has_success = "success" in response_data
|
|
has_data_or_error = any(key in response_data for key in ["data", "error", "printers", "message"])
|
|
|
|
return has_success and has_data_or_error
|
|
|
|
@staticmethod
|
|
def normalize_response(response_data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Normalisiert eine Response auf das Standard-Format.
|
|
|
|
Args:
|
|
response_data: Zu normalisierende Response-Daten
|
|
|
|
Returns:
|
|
dict: Normalisierte Response
|
|
"""
|
|
if ResponseValidator.is_valid_response(response_data):
|
|
return response_data
|
|
|
|
# Legacy-Format konvertieren
|
|
if "printers" in response_data and "success" not in response_data:
|
|
return create_printers_response(
|
|
printers=response_data["printers"],
|
|
message=response_data.get("message")
|
|
)
|
|
|
|
# Error-only Format konvertieren
|
|
if "error" in response_data and "success" not in response_data:
|
|
return create_error_response(
|
|
error=response_data["error"],
|
|
details=response_data.get("details")
|
|
)
|
|
|
|
# Fallback für unbekannte Formate
|
|
return create_error_response(
|
|
error="Unbekannte Response-Struktur",
|
|
details="Die Response konnte nicht verarbeitet werden"
|
|
) |