Programmierung und Automatisierung
Dies ist ein umfangreicher Artikel zum Thema Programmierung und Automatisierung, inklusive Prüfungsfragen, Beispiel-Scripts,
Kernkomponenten, Monitoring-Tools und vieles mehr.
Schau die Überschriften an!
In a Nutshell
Automatisierung durch Programmierung entlastet IT-Prozesse, verbessert Effizienz und reduziert Fehler. Dafür werden Skripte, Programmiersprachen und Automatisierungstools gezielt eingesetzt.
Kompakte Fachbeschreibung
Wiederkehrende Aufgaben wie Backups, Benutzerverwaltung oder Deployments lassen sich durch Skriptprogrammierung automatisieren (z.B. Bash, PowerShell, Python). Dazu kommen Zeitplanung (Cron, Task Scheduler), Monitoring, Logging und Sicherheitsprüfungen.
Auch herstellerspezifische Tools nutzen eigene Skriptbausteine (z.B. VMware PowerCLI). Häufig gehört außerdem dazu, bestehenden Code zu analysieren und gezielt anzupassen, um Automatisierung in bestehende Systeme zu integrieren.
Du wirst immer wieder Dateien, die Du erstellst oder vorliegen hast, bearbeiten müssen. Manchmal muss man zahlreiche “Formatter” oder andere Dinge durch ggf. eine Variable ersetzen, oder einen Link suchen und entfernen. Dann helfen kleine und einfache Python-, Bash, oder PowerShell-Scripts.
Prüfungsrelevante Stichpunkte
Programmiersprachen zielgerichtet auswählen
Die Wahl der richtigen Programmiersprache ist entscheidend für den Automatisierungserfolg. Für einfache Systemaufgaben unter Linux eignet sich Bash, während für komplexe Logik und plattformübergreifende Lösungen Python bevorzugt wird. PowerShell ist die Wahl für Windows-Umgebungen mit tiefen API-Integrationen.
Automatisierung reduziert manuelle Eingriffe und Fehler
Automatisierte Prozesse minimieren menschliche Fehlerquellen und sorgen für konsistente Ergebnisse. Wiederkehrende Aufgaben wie Backups, Benutzeranlagen oder Deployments werden zuverlässig und nachvollziehbar ausgeführt.
Skriptsprachen für Systemaufgaben und Prozesslogik
Skriptsprachen ermöglichen die Automatisierung von Systemaufgaben (Dateiverwaltung, Prozesssteuerung) und komplexer Prozesslogik (Bedingungen, Schleifen, Fehlerbehandlung). Sie bieten schnelle Entwicklung und einfache Anpassungsmöglichkeiten.
Plattformtools: Bash (Linux), PowerShell (Windows)
Bash ist der Standard unter Linux für Shell-Scripting und Systemautomatisierung. PowerShell bietet objektbasierte Verarbeitung und tiefe Integration in Windows-Systeme und APIs. Beide sind IHK-relevant für Systemadministratoren.
Überwachung über Logs, Exit-Codes, Trigger
Automatisierte Prozesse müssen überwacht werden. Logging dokumentiert alle Aktionen, Exit-Codes zeigen Erfolg/Misserfolg an, Trigger ermöglichen ereignisgesteuerte Automatisierung. Dies sorgt für Transparenz und Fehlersicherheit.
Sensible Prozesse nur mit Logging und Rollenprüfung
Bei sensiblen Automatisierungen (z.B. Benutzerrechte, Systemänderungen) sind Logging und Rollenprüfung unerlässlich. Nur autorisierte Benutzer dürfen kritische Skripte ausführen, und alle Aktionen müssen protokolliert werden.
Weniger Aufwand bei Wartung und Systempflege
Automatisierung reduziert den manuellen Aufwand bei Wartung und Systempflege erheblich. Einmal entwickelte Skripte können vielfach eingesetzt und leicht angepasst werden, was Zeit und Ressourcen spart.
Änderungen dokumentieren und versionieren
Alle Änderungen an Automatisierungsskripten müssen dokumentiert und versioniert werden. Dies gewährleistet Nachvollziehbarkeit, ermöglicht Rollbacks und unterstützt die Zusammenarbeit im Team.
Kernkomponenten
1. Wahl der Programmiersprache
Die Auswahl der richtigen Programmiersprache hängt von der Aufgabe und Plattform ab. Python eignet sich für komplexe Logik und plattformübergreifende Lösungen, Bash für Linux-Systemaufgaben, PowerShell für Windows-Umgebungen.
2. Plattformabhängige Skripting-Tools
Jede Plattform bietet spezifische Automatisierungstools: Bash unter Linux, PowerShell unter Windows, Cron für Zeitplanung unter Linux, Task Scheduler unter Windows. Diese Tools ermöglichen direkte Systemintegration.
3. Automatisierungsziele definieren
Vor der Implementierung müssen klare Ziele definiert werden: Welche Aufgaben sollen automatisiert werden? Welche Effizienzgewinne werden erwartet? Welche Sicherheitsanforderungen gelten?
4. Bedingte Ausführung und Schleifen
Automatisierung erfordert oft bedingte Logik (if/else) und Schleifen (for/while) für wiederkehrende Prozesse. Diese Kontrollstrukturen ermöglichen flexible und adaptive Automatisierung.
5. Fehlerbehandlung und Logging
Robuste Automatisierung benötigt umfassende Fehlerbehandlung (try/catch, Exit-Codes) und detailliertes Logging. Dies sichert die Stabilität und Nachvollziehbarkeit automatisierter Prozesse.
6. Zeitgesteuerte Aufgaben (Cron, Task Scheduler)
Cron unter Linux und Task Scheduler unter Windows ermöglichen zeitgesteuerte Ausführung von Skripten. Dies ist ideal für regelmäßige Wartungsaufgaben wie Backups oder Bereinigung.
7. Integration in bestehende Systeme
Automatisierungsskripte müssen in bestehende Systeme integriert werden. Dies betrifft APIs, Datenbanken, Dateisysteme und andere IT-Infrastrukturkomponenten.
8. Versionsverwaltung der Skripte
Alle Automatisierungsskripte sollten unter Versionskontrolle (Git) stehen. Dies ermöglicht Nachverfolgung von Änderungen, Zusammenarbeit und sichere Rollbacks.
9. Sicherheitsprüfungen und Rechtekonzepte
Automatisierung erfordert strenge Sicherheitskonzepte: Rechteprüfung, Rollenmanagement, sichere Passwortverarbeitung und Audit-Logging für sensible Operationen.
10. Monitoring automatisierter Abläufe
Automatisierte Prozesse müssen überwacht werden: Erfolgskontrolle, Fehlerbenachrichtigung, Performance-Monitoring und Alerting bei Abweichungen.
Einfaches Praxisbeispiel (Python-Backup)
import shutil
shutil.copy('/home/user/data.db', '/backup/data.db')Erklärung: Das Skript kopiert automatisiert eine Datei in ein Backup-Verzeichnis und kann regelmäßig per Cronjob laufen.
Warum shutil statt normaler copy-Befehle?
shutil steht für “shell utilities” (nicht shadow utility) und ist ein Python-Standardmodul für hochlevelige Dateioperationen. Es bietet mehrere Vorteile gegenüber einfachen copy-Befehlen:
Vorteile von shutil.copy():
- Plattformübergreifend: Funktioniert auf Windows, Linux und macOS identisch
- Fehlerbehandlung: Wirft saubere Exceptions bei Problemen (Permission, nicht existierende Pfade)
- Metadaten: Erhält Datei-Permissions und Metadaten (bei
shutil.copy2()) - Sicherer: Prüft automatisch auf Overwrite-Probleme
Alternativen und warum sie schlechter sind:
# ❌ Schlecht: OS-spezifisch
import os
os.system('cp /home/user/data.db /backup/data.db') # Nur Linux
os.system('copy C:\\data.db C:\\backup\\data.db') # Nur Windows
# ❌ Schlecht: Keine Fehlerbehandlung
f = open('/home/user/data.db', 'rb')
data = f.read()
f.close()
f = open('/backup/data.db', 'wb')
f.write(data)
f.close()
# ✅ Gut: shutil.copy()
import shutil
shutil.copy('/home/user/data.db', '/backup/data.db')shutil.copy() vs shutil.copy2():
copy(): Kopiert Inhalt + Basis-Permissionscopy2(): Kopiert Inhalt + alle Metadaten (Timestamps, Permissions) - meistens die bessere Wahl für Backups
Vorteile und Nachteile
Vorteile
- Fehlervermeidung durch automatisierte Prozesse
- Zeitersparnis bei wiederkehrenden Aufgaben
- Nachvollziehbarkeit durch Logs
- Plattformübergreifend mit portablen Skriptsprachen
Nachteile
- Fehler im Skript können großen Schaden verursachen
- Erfordert Tests und sauberes Logging
- Unterschiedliches Verhalten je nach Umgebung
Typische Prüfungsfragen (mit Kurzantwort)
- Wann Bash statt Python? Für einfache Systemaufgaben unter Linux (Dateien, Prozesse, Pipes).
- Typische Automatisierungsaufgaben? Backups, User-Anlage, Log-Archivierung, Deployment.
- Wie gestaltet man Automatisierung sicher? Logging, Rechteprüfung, Code Reviews, Exit-Status auswerten.
- Was ist ein Cronjob? Zeitgesteuerter Task unter Linux.
- Vorteile PowerShell gegenüber Bash? Objektbasiert und tief in Windows/APIs integriert.
Freie Antwort
Das Thema verbindet Sprachauswahl, Plattformabhängigkeit und Automatisierungsziel. In Prüfungen ist häufig relevant, ein Skript zu lesen, zu erweitern und Fehlerbehandlung/Logging korrekt einzubauen.
Lernstrategie für dieses Thema
1. Verständniseinstieg: Automatisiere ein kleines Backup lokal
Lösung:
# Bash Backup-Skript
#!/bin/bash
SOURCE="/home/user/documents"
TARGET="/backup/documents_$(date +%Y%m%d)"
mkdir -p "$TARGET"
cp -r "$SOURCE"/* "$TARGET/"
echo "Backup abgeschlossen: $TARGET" >> /var/log/backup.logCron einrichten: 0 2 * * * /pfad/zum/backup.sh
2. Vertiefungsmethode: Skizziere einen Automatisierungsprozess (User-Anlage + Rechte)
Lösung:
# Python User-Anlage mit Rechteprüfung
import subprocess
import logging
logging.basicConfig(filename='user_management.log', level=logging.INFO)
def create_user(username, groups):
try:
# User erstellen
subprocess.run(['useradd', '-m', username], check=True)
# Gruppen zuweisen
for group in groups:
subprocess.run(['usermod', '-aG', group, username], check=True)
logging.info(f"User {username} erstellt mit Gruppen: {groups}")
print(f"User {username} erfolgreich erstellt")
except subprocess.CalledProcessError as e:
logging.error(f"Fehler bei User-Erstellung {username}: {e}")
print(f"Fehler: {e}")
# Verwendung
create_user("newuser", ["sudo", "developers"])3. Prüfungsfokustraining: Skripte analysieren und ergänzen
Lösung:
- Analyse: Skript auf Fehlerbehandlung, Logging und Sicherheit prüfen
- Ergänzung: Exit-Codes auswerten, Parameter validieren, Fehlermeldungen verbessern
- Beispiel-Verbesserung:
import sys
import os
def validate_path(path):
if not os.path.exists(path):
print(f"Fehler: Pfad {path} existiert nicht")
sys.exit(1)
return True
# Verwendung
if len(sys.argv) < 2:
print("Verwendung: skript.py <pfad>")
sys.exit(1)
validate_path(sys.argv[1])
# Rest des Skripts...4. Fehlervermeidung: Logging und Rechteprüfung einbauen
Lösung:
import logging
import os
import sys
from pathlib import Path
# Logging konfigurieren
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('/var/log/automation.log'),
logging.StreamHandler()
]
)
def check_permissions(required_uid=0):
"""Prüft ob Skript mit richtigen Rechten läuft"""
if os.geteuid() != required_uid:
logging.error(f"Skript muss als UID {required_uid} ausgeführt werden")
sys.exit(1)
def safe_operation(operation_func):
"""Decorator für sichere Operationen"""
def wrapper(*args, **kwargs):
try:
result = operation_func(*args, **kwargs)
logging.info(f"Operation {operation_func.__name__} erfolgreich")
return result
except Exception as e:
logging.error(f"Fehler in {operation_func.__name__}: {e}")
raise
return wrapper
@safe_operation
def critical_operation():
# Kritische Operation hier
passThemenanalyse
- Technischer Kern: Shell, Python, Scheduling, Logging
- Implementierungsherausforderungen: Plattformunterschiede, Fehlervermeidung
- Sicherheitsimplikationen: Rechtevergabe, Missbrauch ungesicherter Skripte
- Dokumentationspflichten: Versionskontrolle, Änderungsprotokoll
- Wirtschaftliche Bewertung: weniger manueller Aufwand durch Wiederholbarkeit
Aktuelle Softwareprodukte
Automatisierungs-Plattformen
- Ansible: Configuration Management und Automatisierung (Open Source)
- Jenkins: CI/CD-Pipeline mit extensiver Plugin-Unterstützung
- GitHub Actions: Cloud-native CI/CD direkt im Repository
- GitLab CI/CD: Integrierte Pipeline-Lösung mit Auto DevOps
Skript-Tools
- Visual Studio Code: Moderner Editor mit Python/PowerShell-Unterstützung
- PyCharm: Python-IDE mit Debugging und Testing-Funktionen
- Windows Terminal: Moderne Terminal-Umgebung für PowerShell
- WSL (Windows Subsystem for Linux): Linux-Umgebung unter Windows
Monitoring und Logging
- Prometheus: Metrik-Sammlung und Alerting
- Grafana: Visualisierung von Monitoring-Daten
- ELK Stack: Elasticsearch, Logstash, Kibana für Log-Analyse
- Nagios: klassisches Monitoring-System
Cloud-Automatisierung
- Azure Automation: PowerShell/Python Runbooks in Azure
- AWS Lambda: Serverless Functions für Automatisierung
- Google Cloud Functions: Event-gesteuerte Automatisierung
- Terraform: Infrastructure as Code für Multi-Cloud
Fragen und Antworten
Frage 1: Welche Sprache für welche Automatisierungsaufgabe?
Antwort: Bash für einfache Linux-Systemaufgaben, PowerShell für Windows-Systemintegration, Python für komplexe Logik und plattformübergreifende Lösungen.
Frage 2: Wie sichert man Automatisierungsskripte?
Antwort: Versionskontrolle mit Git, Code Reviews, umfassende Fehlerbehandlung, Logging, Rechteprüfung und regelmäßige Security-Audits.
Frage 3: Was sind typische Fehler bei der Automatisierung?
Antwort: Fehlende Fehlerbehandlung, unzureichendes Logging, harte Codierung von Pfaden, mangelnde Tests, ignorieren von Exit-Codes.
Frage 4: Wie testet man Automatisierungsskripte?
Antwort: Unit-Tests für Funktionen, Integration-Tests für Workflows, Staging-Umgebungen für Production-Tests, Monitoring im Live-Betrieb.
Frage 5: Welche Alternativen zu Cron gibt es?
Antwort: systemd-Timer (modernere Alternative), Jenkins Scheduled Jobs, Kubernetes CronJobs, Cloud-spezifische Scheduler (AWS EventBridge).
Weiterführende Infos
- http://linuxcommand.org/lc3_learning_the_shell.php
- https://learn.microsoft.com/de-de/powershell/
- https://automatetheboringstuff.com/
- https://docs.github.com/en/actions
- https://crontab.guru/
Buchempfehlung
Vertiefe Dein Wissen über Python-Automatisierung mit diesem praxisorientierten Handbuch für Einsteiger und Fortgeschrittene. Dieses Buch bietet konkrete Lösungen für reale Automatisierungsaufgaben und hilft Dir, effiziente Skripte zu entwickeln.
Routineaufgaben mit Python automatisieren: Praktische Programmierlösungen für Einsteiger*innenDieses Buch ist ideal für:
- Einsteiger, die Python für Automatisierung lernen wollen
- Systemadministratoren, die wiederkehrende Aufgaben automatisieren möchten
- Entwickler, die ihre täglichen Workflows optimieren wollen
- Alle, die praktische Python-Lösungen für alltägliche Probleme suchen
Kleine Script-Sammlung an kleinen (Windows/Linux/Mac) Automatisierungen
Hier sind 15 praktische Python-Skripte für typische Automatisierungsaufgaben auf den verschiedenen Plattformen.
Linux-Skripte (5)
1. Bilder verkleinern und optimieren
from PIL import Image
import os
from pathlib import Path
def resize_images(folder_path, max_size=1920, quality=85):
"""Verkleinert alle Bilder in einem Ordner"""
for file in Path(folder_path).glob('*'):
if file.suffix.lower() in ['.jpg', '.jpeg', '.png']:
with Image.open(file) as img:
img.thumbnail((max_size, max_size))
img.save(file, quality=quality, optimize=True)
print(f"Optimiert: {file.name}")
# Nutzung: resize_images('/home/user/Bilder')2. Dubletten in einem Ordner finden
import hashlib
from pathlib import Path
from collections import defaultdict
def find_duplicates(folder_path):
"""Findet doppelte Dateien basierend auf MD5-Hash"""
hashes = defaultdict(list)
for file in Path(folder_path).rglob('*'):
if file.is_file():
file_hash = hashlib.md5(file.read_bytes()).hexdigest()
hashes[file_hash].append(file)
duplicates = {h: files for h, files in hashes.items() if len(files) > 1}
for hash_val, files in duplicates.items():
print(f"Duplikate gefunden ({len(files)} Dateien):")
for file in files:
print(f" - {file}")
# Nutzung: find_duplicates('/home/user/Downloads')3. Log-Dateien automatisch bereinigen
import os
from pathlib import Path
from datetime import datetime, timedelta
def clean_old_logs(log_folder, days=30):
"""Löscht Log-Dateien älter als X Tage"""
cutoff_date = datetime.now() - timedelta(days=days)
for file in Path(log_folder).glob('*.log'):
if datetime.fromtimestamp(file.stat().st_mtime) < cutoff_date:
file.unlink()
print(f"Gelöscht: {file.name}")
# Nutzung: clean_old_logs('/var/log/myapp', days=30)4. PDF-Dateien zusammenfügen
from PyPDF2 import PdfMerger
from pathlib import Path
def merge_pdfs(folder_path, output_name='combined.pdf'):
"""Fügt alle PDFs in einem Ordner zusammen"""
merger = PdfMerger()
for file in sorted(Path(folder_path).glob('*.pdf')):
merger.append(file)
print(f"Hinzugefügt: {file.name}")
merger.write(output_name)
merger.close()
print(f"Erstellt: {output_name}")
# Nutzung: merge_pdfs('/home/user/Dokumente/PDFs')5. System-Überwachung und Alert
import psutil
import smtplib
from email.mime.text import MIMEText
def check_disk_usage(threshold=90):
"""Prüft Festplattenbelegung und sendet Alert bei Überschreitung"""
for partition in psutil.disk_partitions():
usage = psutil.disk_usage(partition.mountpoint)
percent = usage.percent
if percent > threshold:
print(f"ALARM: {partition.mountpoint} ist {percent}% voll")
# Hier könnte eine E-Mail gesendet werden
# send_alert_email(partition.mountpoint, percent)
# Nutzung: check_disk_usage(threshold=90)Windows-Skripte (5)
1. Desktop aufräumen
import shutil
from pathlib import Path
from datetime import datetime
def clean_downloads():
"""Sortiert Dateien im Downloads-Ordner"""
downloads = Path.home() / 'Downloads'
# Ordner erstellen
folders = {
'Bilder': ['.jpg', '.jpeg', '.png', '.gif'],
'Dokumente': ['.pdf', '.doc', '.docx', '.txt'],
'Musik': ['.mp3', '.wav', '.flac'],
'Videos': ['.mp4', '.avi', '.mkv'],
'Archiv': ['.zip', '.rar', '.7z']
}
for folder, extensions in folders.items():
target = downloads / folder
target.mkdir(exist_ok=True)
for file in downloads.glob('*'):
if file.suffix.lower() in extensions and file.is_file():
shutil.move(str(file), str(target / file.name))
print(f"Verschoben: {file.name} -> {folder}")
# Nutzung: clean_downloads()2. Temporäre Dateien löschen
import shutil
import os
from pathlib import Path
def clean_temp_files():
"""Löscht temporäre Dateien und Cache"""
temp_folders = [
Path(os.environ.get('TEMP', '')),
Path(os.environ.get('TMP', '')),
Path.home() / 'AppData' / 'Local' / 'Temp',
]
for folder in temp_folders:
if folder.exists():
try:
for file in folder.glob('*'):
if file.is_file():
file.unlink()
print(f"Bereinigt: {folder}")
except PermissionError:
print(f"Kein Zugriff: {folder}")
# Nutzung: clean_temp_files()3. WLAN-Netzwerke scannen
import subprocess
def scan_wifi_networks():
"""Zeigt verfügbare WLAN-Netzwerke an"""
try:
result = subprocess.run(
['netsh', 'wlan', 'show', 'networks'],
capture_output=True,
text=True
)
networks = []
for line in result.stdout.split('\n'):
if 'SSID' in line:
ssid = line.split(':')[1].strip()
networks.append(ssid)
print("Verfügbare Netzwerke:")
for network in networks:
print(f" - {network}")
except Exception as e:
print(f"Fehler: {e}")
# Nutzung: scan_wifi_networks()4. Clipboard-Manager
import pyperclip
import time
from datetime import datetime
def clipboard_history(max_entries=10):
"""Speichert Clipboard-Historie"""
history = []
last_content = ""
print("Clipboard-Überwachung gestartet (Strg+C zum Beenden)")
try:
while True:
current = pyperclip.paste()
if current != last_content and current:
timestamp = datetime.now().strftime("%H:%M:%S")
history.insert(0, f"[{timestamp}] {current[:50]}...")
if len(history) > max_entries:
history.pop()
print(f"\nNeuer Eintrag: {current[:30]}...")
for entry in history:
print(f" {entry}")
last_content = current
time.sleep(2)
except KeyboardInterrupt:
print("\nClipboard-Überwachung beendet")
# Nutzung: clipboard_history()5. Screenshot bei Tastendruck
import pyautogui
import keyboard
from datetime import datetime
from pathlib import Path
def screenshot_on_hotkey(hotkey='f9'):
"""Macht Screenshot bei Tastendruck"""
screenshots = Path.home() / 'Desktop' / 'Screenshots'
screenshots.mkdir(exist_ok=True)
def take_screenshot():
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = screenshots / f"screenshot_{timestamp}.png"
pyautogui.screenshot(str(filename))
print(f"Screenshot gespeichert: {filename}")
keyboard.add_hotkey(hotkey, take_screenshot)
print(f"Drücke {hotkey.upper()} für Screenshot")
keyboard.wait() # Wartet auf Tastendruck
# Nutzung: screenshot_on_hotkey('f9')Mac-Skripte (5)
1. Finder-Sortierung automatisieren
import shutil
from pathlib import Path
def organize_downloads():
"""Sortiert Downloads-Ordner auf Mac"""
downloads = Path.home() / 'Downloads'
file_types = {
'Bilder': ['.jpg', '.jpeg', '.png', '.heic', '.gif'],
'Dokumente': ['.pdf', '.pages', '.doc', '.docx', '.txt'],
'Musik': ['.mp3', '.m4a', '.wav', '.aiff'],
'Videos': ['.mp4', '.mov', '.avi', '.mkv'],
'Archiv': ['.zip', '.rar', '.7z', '.dmg']
}
for folder, extensions in file_types.items():
target = downloads / folder
target.mkdir(exist_ok=True)
for file in downloads.glob('*'):
if file.suffix.lower() in extensions and file.is_file():
shutil.move(str(file), str(target / file.name))
print(f"Organisiert: {file.name}")
# Nutzung: organize_downloads()2. macOS-Notifications senden
import subprocess
def send_notification(title, message):
"""Sendet macOS-Systembenachrichtigung"""
cmd = f"""
osascript -e 'display notification "{message}" with title "{title}"'
"""
subprocess.run(cmd, shell=True)
# Nutzung: send_notification("Backup fertig", "Dein Backup wurde erfolgreich abgeschlossen")3. Spotify steuern
import subprocess
def spotify_control(action):
"""Steuert Spotify über AppleScript"""
commands = {
'play': 'tell application "Spotify" to play',
'pause': 'tell application "Spotify" to pause',
'next': 'tell application "Spotify" to next track',
'previous': 'tell application "Spotify" to previous track'
}
if action in commands:
cmd = f'osascript -e \'tell application "Spotify" to {commands[action]}\''
subprocess.run(cmd, shell=True)
print(f"Spotify: {action}")
else:
print(f"Unbekannter Befehl: {action}")
# Nutzung: spotify_control('play')4. Bildschirmhelligkeit steuern
import subprocess
def set_brightness(level):
"""Setzt Bildschirmhelligkeit (0-100)"""
if 0 <= level <= 100:
cmd = f"""
osascript -e 'tell application "System Events" to set brightness of every display to {level/100}'
"""
subprocess.run(cmd, shell=True)
print(f"Helligkeit auf {level}% gesetzt")
else:
print("Helligkeit muss zwischen 0 und 100 liegen")
# Nutzung: set_brightness(50)5. PDF-Text extrahieren
import PyPDF2
from pathlib import Path
def extract_pdf_text(pdf_path):
"""Extrahiert Text aus PDF-Datei"""
text = ""
with open(pdf_path, 'rb') as file:
reader = PyPDF2.PdfReader(file)
for page in reader.pages:
text += page.extract_text() + "\n"
return text
def save_pdf_text_to_file(pdf_path, output_path):
"""Speichert extrahierten Text in Datei"""
text = extract_pdf_text(pdf_path)
with open(output_path, 'w', encoding='utf-8') as file:
file.write(text)
print(f"Text extrahiert: {output_path}")
# Nutzung: save_pdf_text_to_file('/Users/user/Dokument.pdf', '/Users/user/Dokument.txt')Hinweise zur Nutzung
-
Installiere benötigte Pakete:
pip install pillow pyperclip pyautogui pypdf2 psutil -
Rechte beachten: Einige Skripte benötigen Administrator-Rechte
-
Pfade anpassen: Pfade an deine lokale Umgebung anpassen
-
Testen: Skripte immer zuerst mit Testdaten ausprobieren
-
Sicherheitskopien: Vor dem Ausführen von Lösch-Operationen Backups erstellen
Bash und PowerShell Script-Sammlung
Ja, Mac hat standardmäßig Bash (oder zsh als Default in neueren macOS-Versionen). Hier sind 15 weitere Skripte in den nativen Shell-Sprachen.
Linux Bash-Skripte (5)
1. Automatisches Backup mit Rotation
#!/bin/bash
# backup.sh - Automatisches Backup mit Rotation
SOURCE_DIR="/home/user/documents"
BACKUP_DIR="/backup"
MAX_BACKUPS=7
DATE=$(date +%Y%m%d_%H%M%S)
BACKUP_NAME="backup_$DATE.tar.gz"
# Backup erstellen
tar -czf "$BACKUP_DIR/$BACKUP_NAME" "$SOURCE_DIR"
echo "Backup erstellt: $BACKUP_NAME"
# Alte Backups löschen (Rotation)
cd "$BACKUP_DIR"
ls -t backup_*.tar.gz | tail -n +$((MAX_BACKUPS + 1)) | xargs -r rm
echo "Alte Backups bereinigt (max $MAX_BACKUPS behalten)"2. Server-Status-Check
#!/bin/bash
# server_check.sh - Prüft Server-Status und Services
echo "=== Server Status Check ==="
echo "Zeit: $(date)"
echo ""
# CPU-Last
echo "CPU-Last:"
top -bn1 | grep "Cpu(s)" | sed "s/.*, *\([0-9.]*\)%id.*/\1/" | awk '{print 100 - $1"%"}'
# Speicher
echo ""
echo "Speicher:"
free -h
# Festplatte
echo ""
echo "Festplatte:"
df -h
# Wichtige Services prüfen
echo ""
echo "Services:"
for service in nginx mysql docker; do
if systemctl is-active --quiet "$service"; then
echo "✓ $service läuft"
else
echo "✗ $service läuft nicht"
fi
done3. Log-Dateien analysieren
#!/bin/bash
# log_analyzer.sh - Analysiert Log-Dateien auf Fehler
LOG_FILE="/var/log/syslog"
ERROR_COUNT=$(grep -i "error" "$LOG_FILE" | wc -l)
WARNING_COUNT=$(grep -i "warning" "$LOG_FILE" | wc -l)
echo "=== Log-Analyse für $LOG_FILE ==="
echo "Fehler: $ERROR_COUNT"
echo "Warnungen: $WARNING_COUNT"
echo ""
# Letzte 10 Fehler anzeigen
if [ $ERROR_COUNT -gt 0 ]; then
echo "Letzte 10 Fehler:"
grep -i "error" "$LOG_FILE" | tail -n 10
fi4. Dateien nach Größe finden
#!/bin/bash
# find_large_files.sh - Findet große Dateien
DIRECTORY="/home/user"
MIN_SIZE="100M" # Mindestgröße
echo "=== Dateien größer als $MIN_SIZE in $DIRECTORY ==="
find "$DIRECTORY" -type f -size +"$MIN_SIZE" -exec ls -lh {} \; | awk '{print $5, $9}'5. Prozess-Monitoring
#!/bin/bash
# process_monitor.sh - Überwacht spezifische Prozesse
PROCESSES=("nginx" "mysql" "redis" "docker")
echo "=== Prozess-Monitoring ==="
for process in "${PROCESSES[@]}"; do
if pgrep -x "$process" > /dev/null; then
PID=$(pgrep -x "$process")
MEMORY=$(ps -p "$PID" -o %mem --no-headers)
CPU=$(ps -p "$PID" -o %cpu --no-headers)
echo "✓ $process (PID: $PID, RAM: $MEMORY%, CPU: $CPU%)"
else
echo "✗ $process läuft nicht"
fi
doneWindows PowerShell-Skripte (5)
1. System-Informationen sammeln
# system_info.ps1 - Sammelt detaillierte System-Informationen
Write-Host "=== System-Informationen ===" -ForegroundColor Cyan
# System-Info
Write-Host "`nSystem:" -ForegroundColor Yellow
Get-CimInstance Win32_ComputerSystem | Select-Object Manufacturer, Model, TotalPhysicalMemory
# Betriebssystem
Write-Host "`nBetriebssystem:" -ForegroundColor Yellow
Get-CimInstance Win32_OperatingSystem | Select-Object Caption, Version, BuildNumber
# CPU
Write-Host "`nCPU:" -ForegroundColor Yellow
Get-CimInstance Win32_Processor | Select-Object Name, NumberOfCores, MaxClockSpeed
# Festplatten
Write-Host "`nFestplatten:" -ForegroundColor Yellow
Get-CimInstance Win32_LogicalDisk | Select-Object DeviceID, Size, FreeSpace | Format-Table2. User-Accounts auflisten
# user_accounts.ps1 - Listet alle User-Accounts auf
Write-Host "=== User-Accounts ===" -ForegroundColor Cyan
Get-LocalUser | Select-Object Name, Enabled, LastLogon | Format-Table
# Admin-Accounts
Write-Host "`nAdmin-Accounts:" -ForegroundColor Yellow
Get-LocalGroupMember -Group "Administrators" | Select-Object Name, PrincipalSource3. Windows-Updates prüfen
# check_updates.ps1 - Prüft auf Windows-Updates
Write-Host "=== Windows-Updates prüfen ===" -ForegroundColor Cyan
try {
$UpdateSession = New-Object -ComObject Microsoft.Update.Session
$UpdateSearcher = $UpdateSession.CreateUpdateSearcher()
$Updates = $UpdateSearcher.Search("IsInstalled=0").Updates
if ($Updates.Count -eq 0) {
Write-Host "Keine ausstehenden Updates" -ForegroundColor Green
} else {
Write-Host "$($Updates.Count) ausstehende Updates:" -ForegroundColor Yellow
foreach ($Update in $Updates) {
Write-Host " - $($Update.Title)"
}
}
} catch {
Write-Host "Fehler beim Prüfen der Updates: $_" -ForegroundColor Red
}4. Netzwerk-Verbindungen überwachen
# network_monitor.ps1 - Überwacht aktive Netzwerk-Verbindungen
Write-Host "=== Aktive Netzwerk-Verbindungen ===" -ForegroundColor Cyan
Get-NetTCPConnection | Where-Object {$_.State -eq "Established"} |
Select-Object LocalAddress, LocalPort, RemoteAddress, RemotePort, State |
Format-Table
# Offene Ports
Write-Host "`nOffene Ports:" -ForegroundColor Yellow
Get-NetTCPConnection | Where-Object {$_.State -eq "Listening"} |
Select-Object LocalAddress, LocalPort, OwningProcess |
Format-Table5. Event-Log analysieren
# event_log.ps1 - Analysiert Windows Event-Logs
Write-Host "=== Event-Log Analyse (letzte 24h) ===" -ForegroundColor Cyan
$StartDate = (Get-Date).AddDays(-1)
# Fehler im System-Log
Write-Host "`nSystem-Fehler:" -ForegroundColor Yellow
Get-EventLog -LogName System -EntryType Error -After $StartDate |
Select-Object TimeGenerated, Source, Message |
Format-Table -Wrap
# Warnungen im Application-Log
Write-Host "`nApplication-Warnungen:" -ForegroundColor Yellow
Get-EventLog -LogName Application -EntryType Warning -After $StartDate |
Select-Object TimeGenerated, Source, Message |
Format-Table -WrapMac Bash/zsh-Skripte (5)
1. Mac-System-Info
#!/bin/bash
# mac_info.sh - Zeigt Mac-System-Informationen
echo "=== Mac-System-Informationen ==="
echo ""
# Hardware
echo "Hardware:"
system_profiler SPHardwareDataType | grep -E "Model Name|Processor|Memory"
# macOS-Version
echo ""
echo "macOS-Version:"
sw_vers
# Festplatte
echo ""
echo "Festplatte:"
df -h
# Battery-Status (Laptops)
echo ""
echo "Battery-Status:"
system_profiler SPPowerDataType | grep -E "Charge|Capacity"2. Homebrew-Pakete aktualisieren
#!/bin/bash
# brew_update.sh - Aktualisiert Homebrew-Pakete
echo "=== Homebrew Update ==="
echo ""
# Homebrew selbst aktualisieren
echo "Homebrew aktualisieren..."
brew update
# Pakete aktualisieren
echo ""
echo "Pakete aktualisieren..."
brew upgrade
# Bereinigung
echo ""
echo "Bereinigung..."
brew cleanup
echo "Update abgeschlossen!"3. Finder-Sortierung automatisieren
#!/bin/bash
# finder_organize.sh - Sortiert Downloads-Ordner
DOWNLOADS="$HOME/Downloads"
echo "=== Downloads organisieren ==="
# Ordner erstellen
mkdir -p "$DOWNLOADS"/{Bilder,Dokumente,Musik,Videos,Archiv}
# Dateien verschieben
for file in "$DOWNLOADS"/*; do
if [ -f "$file" ]; then
case "${file,,}" in
*.jpg|*.jpeg|*.png|*.gif|*.heic)
mv "$file" "$DOWNLOADS/Bilder/"
echo "Bilder: $(basename "$file")"
;;
*.pdf|*.doc|*.docx|*.txt|*.pages)
mv "$file" "$DOWNLOADS/Dokumente/"
echo "Dokumente: $(basename "$file")"
;;
*.mp3|*.m4a|*.wav|*.aiff)
mv "$file" "$DOWNLOADS/Musik/"
echo "Musik: $(basename "$file")"
;;
*.mp4|*.mov|*.avi|*.mkv)
mv "$file" "$DOWNLOADS/Videos/"
echo "Videos: $(basename "$file")"
;;
*.zip|*.rar|*.7z|*.dmg)
mv "$file" "$DOWNLOADS/Archiv/"
echo "Archiv: $(basename "$file")"
;;
esac
fi
done
echo "Organisation abgeschlossen!"4. SSH-Keys verwalten
#!/bin/bash
# ssh_keys.sh - Verwaltet SSH-Keys
SSH_DIR="$HOME/.ssh"
echo "=== SSH-Keys verwalten ==="
echo ""
# SSH-Verzeichnis prüfen
if [ ! -d "$SSH_DIR" ]; then
echo "SSH-Verzeichnis erstellen..."
mkdir -p "$SSH_DIR"
chmod 700 "$SSH_DIR"
fi
# Vorhandene Keys auflisten
echo "Vorhandene Keys:"
ls -la "$SSH_DIR"/*.pub 2>/dev/null || echo "Keine öffentlichen Keys gefunden"
# Neuen Key erstellen (optional)
read -p "Neuen SSH-Key erstellen? (j/n): " create_key
if [ "$create_key" = "j" ]; then
read -p "Key-Name (z.B. github): " key_name
ssh-keygen -t ed25519 -f "$SSH_DIR/$key_name" -C "$key_name"
echo "Key erstellt: $SSH_DIR/$key_name"
fi5. Dock-Backup und Restore
#!/bin/bash
# dock_backup.sh - Sichert und wiederherstellt Dock-Konfiguration
DOCK_PLIST="$HOME/Library/Preferences/com.apple.dock.plist"
BACKUP_DIR="$HOME/DockBackups"
echo "=== Dock-Backup ==="
# Backup erstellen
backup_dock() {
mkdir -p "$BACKUP_DIR"
DATE=$(date +%Y%m%d_%H%M%S)
cp "$DOCK_PLIST" "$BACKUP_DIR/dock_backup_$DATE.plist"
echo "Dock gesichert: dock_backup_$DATE.plist"
}
# Dock wiederherstellen
restore_dock() {
echo "Verfügbare Backups:"
ls -lt "$BACKUP_DIR"/*.plist
read -p "Backup-Datei eingeben: " backup_file
cp "$backup_file" "$DOCK_PLIST"
killall Dock
echo "Dock wiederhergestellt und neu gestartet"
}
# Menü
echo "1: Dock sichern"
echo "2: Dock wiederherstellen"
read -p "Auswahl: " choice
case $choice in
1) backup_dock ;;
2) restore_dock ;;
*) echo "Ungültige Auswahl" ;;
esacHinweise zur Shell-Skript-Nutzung
Bash-Skripte (Linux/Mac):
- Ausführbar machen:
chmod +x skript.sh - Ausführen:
./skript.shoderbash skript.sh - Pfad prüfen:
which bash(meist/bin/bash)
PowerShell-Skripte (Windows):
- Execution-Policy prüfen:
Get-ExecutionPolicy - Falls nötig ändern:
Set-ExecutionPolicy RemoteSigned - Ausführen:
.\skript.ps1oderpowershell -File skript.ps1
Mac-spezifisch:
- Standard-Shell ist zsh (seit macOS Catalina)
- Bash ist weiterhin verfügbar:
/bin/bash - Für AppleScript-Integration:
osascriptverwenden
Best Practices für Automatisierungsskripte
1. Coding Standards und Namenskonventionen
Python:
# ✅ Gute Namenskonventionen
def backup_database(source_path: str, backup_dir: str) -> bool:
"""Erstellt Backup der Datenbank"""
pass
# ❌ Schlechte Namenskonventionen
def db(s, d):
passBash:
# ✅ Gute Namenskonventionen
backup_database() {
local source_path="$1"
local backup_dir="$2"
}
# ❌ Schlechte Namenskonventionen
bd() {
s=$1
d=$2
}PowerShell:
# ✅ Gute Namenskonventionen (PascalCase, Verb-Noun)
function Backup-Database {
param(
[string]$SourcePath,
[string]$BackupDir
)
}
# ❌ Schlechte Namenskonventionen
function bd {
param($s, $d)
}2. Dokumentation und Kommentare
Docstrings und Kommentare:
def process_files(directory: str, pattern: str) -> list:
"""
Verarbeitet alle Dateien in einem Verzeichnis die einem Muster entsprechen.
Args:
directory: Pfad zum Verzeichnis
pattern: Dateimuster (z.B. '*.log')
Returns:
Liste der verarbeiteten Dateien
Raises:
FileNotFoundError: Wenn Verzeichnis nicht existiert
"""
if not os.path.exists(directory):
raise FileNotFoundError(f"Verzeichnis nicht gefunden: {directory}")
# ... Rest der FunktionBash-Comments:
#!/bin/bash
# backup.sh - Automatisches Backup mit Rotation
# Erstellt: 2026-07-14
# Autor: Ihr Name
# Version: 1.0
# Konfiguration
SOURCE_DIR="/home/user/documents" # Quellverzeichnis
BACKUP_DIR="/backup" # Backup-Verzeichnis
MAX_BACKUPS=7 # Max. Anzahl Backups3. Modularisierung und Wiederverwendbarkeit
Funktionen statt Monolith:
# ✅ Modularisiert
def validate_path(path: str) -> bool:
"""Prüft ob Pfad existiert"""
return os.path.exists(path)
def copy_file(source: str, target: str) -> bool:
"""Kopiert Datei mit Fehlerbehandlung"""
try:
shutil.copy(source, target)
return True
except Exception as e:
log_error(f"Kopierfehler: {e}")
return False
def backup_files(files: list, backup_dir: str) -> bool:
"""Hauptfunktion für Backup"""
if not validate_path(backup_dir):
return False
for file in files:
if not copy_file(file, backup_dir):
return False
return True4. Konfiguration von Code trennen
Config-Files verwenden:
# config.yaml
backup:
source_dir: "/home/user/documents"
backup_dir: "/backup"
max_backups: 7
compression: "gzip"
# script.py
import yaml
def load_config(config_file: str) -> dict:
"""Lädt Konfiguration aus YAML-Datei"""
with open(config_file, 'r') as f:
return yaml.safe_load(f)
config = load_config('config.yaml')
backup_config = config['backup']5. Logging und Monitoring
Strukturiertes Logging:
import logging
from datetime import datetime
# Logging konfigurieren
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('automation.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
def process_data(data: str):
"""Verarbeitet Daten mit Logging"""
logger.info(f"Verarbeitung gestartet: {len(data)} Zeichen")
try:
result = transform_data(data)
logger.info(f"Verarbeitung erfolgreich")
return result
except Exception as e:
logger.error(f"Verarbeitung fehlgeschlagen: {e}")
raise6. Fehlerbehandlung und Recovery
Robuste Fehlerbehandlung:
import time
from functools import wraps
def retry(max_attempts: int = 3, delay: int = 1):
"""Decorator für Retry-Mechanismus"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_attempts - 1:
raise
logger.warning(f"Versuch {attempt + 1} fehlgeschlagen: {e}")
time.sleep(delay)
return wrapper
return decorator
@retry(max_attempts=3, delay=2)
def download_file(url: str, target: str):
"""Lädt Datei mit Retry herunter"""
# Download-Logik
pass7. Testing und Validation
Einfache Tests:
def test_validate_path():
"""Test für validate_path Funktion"""
# Test mit existierendem Pfad
assert validate_path('/tmp') == True
# Test mit nicht-existierendem Pfad
assert validate_path('/nicht/existent') == False
print("✓ validate_path Tests bestanden")
if __name__ == '__main__':
test_validate_path()8. Performance-Optimierung
Effiziente Verarbeitung:
# ❌ Ineffizient (für große Dateien)
def process_large_file_slow(filepath: str):
with open(filepath, 'r') as f:
content = f.read() # Lädt alles in RAM
return process_content(content)
# ✅ Effizient (streaming)
def process_large_file_fast(filepath: str):
with open(filepath, 'r') as f:
for line in f: # Zeilenweise Verarbeitung
yield process_line(line)9. Sicherheit bei Skripten
Keine Hardcoding von Secrets:
# ❌ Schlecht - Passworte im Code
password = "geheim123"
# ✅ Gut - Environment Variables
import os
password = os.getenv('DB_PASSWORD')
# ✅ Noch besser - Secrets Manager
from keyring import get_password
password = get_password('system', 'username')10. Versionierung und Deployment
Semantic Versioning:
# Versionierung im Skript
VERSION="1.2.3" # MAJOR.MINOR.PATCH
# Changelog
# 1.2.3 - Bugfix für Backup-Rotation
# 1.2.0 - Neue Logging-Funktion
# 1.0.0 - Erste VersionSicherheit bei Automatisierungsskripten
1. Secrets Management
Environment Variables:
# .env Datei (nicht in Git committen!)
DB_PASSWORD=geheim123
API_KEY=sk-1234567890
# script.py
import os
from dotenv import load_dotenv
load_dotenv() # Lädt .env Datei
db_password = os.getenv('DB_PASSWORD')
api_key = os.getenv('API_KEY')Secrets Manager:
# AWS Secrets Manager
import boto3
def get_secret(secret_name: str) -> str:
"""Holt Secret aus AWS Secrets Manager"""
client = boto3.client('secretsmanager')
response = client.get_secret_value(SecretId=secret_name)
return response['SecretString']
# Azure Key Vault
from azure.keyvault.secrets import SecretClient
from azure.identity import DefaultAzureCredential
def get_azure_secret(vault_url: str, secret_name: str) -> str:
"""Holt Secret aus Azure Key Vault"""
credential = DefaultAzureCredential()
client = SecretClient(vault_url=vault_url, credential=credential)
return client.get_secret(secret_name).value2. Input Validation
Benutzereingaben validieren:
import re
from pathlib import Path
def validate_filename(filename: str) -> bool:
"""Validiert Dateinamen auf Sicherheit"""
# Erlaubt nur alphanumerische Zeichen, Unterstrich, Bindestrich
pattern = r'^[a-zA-Z0-9_-]+\.[a-zA-Z0-9]{3,4}$'
return bool(re.match(pattern, filename))
def validate_path(path: str) -> bool:
"""Validiert Pfad auf Path Traversal Angriffe"""
resolved_path = Path(path).resolve()
# Prüft ob Pfad außerhalb des erlaubten Bereichs liegt
return not any(part in ['..', '~'] for part in resolved_path.parts)
# Nutzung
filename = input("Dateiname: ")
if not validate_filename(filename):
raise ValueError("Ungültiger Dateiname")3. Rechte und Permissions
Minimale Rechte-Prinzip:
#!/bin/bash
# Skript mit minimalen Rechten ausführen
# Prüfen ob Skript als root läuft
if [ "$EUID" -eq 0 ]; then
echo "Dieses Skript sollte nicht als root ausgeführt werden"
exit 1
fi
# Dateien mit sicheren Permissions erstellen
umask 077 # Nur Owner hat Rechte
touch sensitive_file.txtPython-Permissions:
import os
import stat
def set_secure_permissions(filepath: str):
"""Setzt sichere Datei-Permissions (nur Owner)"""
os.chmod(filepath, stat.S_IRUSR | stat.S_IWUSR) # 0600
def set_executable_permissions(filepath: str):
"""Setzt ausführbare Permissions (Owner rwx)"""
os.chmod(filepath, stat.S_IRWXU) # 07004. SQL Injection Prevention
Parameterized Queries:
# ❌ UNSICHER - SQL Injection möglich
query = f"SELECT * FROM users WHERE name = '{username}'"
cursor.execute(query)
# ✅ SICHER - Parameterized Query
query = "SELECT * FROM users WHERE name = %s"
cursor.execute(query, (username,))5. Command Injection Prevention
Sichere Shell-Ausführung:
import subprocess
# ❌ UNSICHER - Command Injection möglich
user_input = "user; rm -rf /"
os.system(f"echo {user_input}")
# ✅ SICHER - subprocess mit Liste
subprocess.run(['echo', user_input], check=True)
# ✅ Noch sicherer - shlex.quote
import shlex
safe_input = shlex.quote(user_input)
subprocess.run(f"echo {safe_input}", shell=True, check=True)6. File Operation Safety
Sichere Dateioperationen:
import os
import tempfile
def safe_file_write(filepath: str, content: str):
"""Sicheres Schreiben mit Atomic Write"""
# Temporäre Datei im gleichen Verzeichnis
dir_path = os.path.dirname(filepath)
with tempfile.NamedTemporaryFile(
mode='w',
dir=dir_path,
delete=False
) as tmp_file:
tmp_file.write(content)
tmp_path = tmp_file.name
# Atomic rename
os.rename(tmp_path, filepath)
def safe_file_delete(filepath: str):
"""Sicheres Löschen mit Bestätigung"""
if not os.path.exists(filepath):
return False
# Bestätigung einholen
response = input(f"Soll {filepath} wirklich gelöscht werden? (j/n): ")
if response.lower() == 'j':
os.remove(filepath)
return True
return False7. Logging von Sensitiven Daten
Keine Secrets in Logs:
import logging
from typing import Any
class SecureFormatter(logging.Formatter):
"""Formatter der sensitive Daten maskiert"""
SENSITIVE_KEYS = ['password', 'token', 'secret', 'key']
def format(self, record: logging.LogRecord) -> str:
msg = super().format(record)
for key in self.SENSITIVE_KEYS:
msg = msg.replace(f'{key}=', f'{key}=***')
return msg
# Nutzung
logger = logging.getLogger(__name__)
handler = logging.StreamHandler()
handler.setFormatter(SecureFormatter())
logger.addHandler(handler)
logger.info("Login mit password=geheim123")
# Ausgabe: Login mit password=***8. HTTPS und TLS
Sichere Netzwerkverbindungen:
import requests
import urllib3
# SSL-Warnungen aktivieren (nicht in Production!)
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# ✅ SICHER - HTTPS mit Zertifikatsprüfung
response = requests.get('https://api.example.com', verify=True)
# ❌ UNSICHER - SSL-Verifikation deaktiviert
response = requests.get('https://api.example.com', verify=False)
# ✅ SICHER - Eigenes CA-Zertifikat
response = requests.get(
'https://api.example.com',
verify='/path/to/ca-bundle.crt'
)9. Dependency Security
Vulnerability Scanning:
# Python Dependencies scannen
pip install safety
safety check
# npm Dependencies scannen
npm audit
# Alle Dependencies aktualisieren
pip install --upgrade pip
pip list --outdated10. Security Audits
Regelmäßige Security-Checks:
import subprocess
import json
def run_security_audit():
"""Führt grundlegende Security-Checks durch"""
checks = {
'python_version': check_python_version,
'dependencies': check_dependencies,
'file_permissions': check_file_permissions,
'hardcoded_secrets': check_hardcoded_secrets
}
results = {}
for check_name, check_func in checks.items():
results[check_name] = check_func()
return results
def check_hardcoded_secrets():
"""Prüft auf hardcoded Secrets im Code"""
import re
secret_patterns = [
r'password\s*=\s*["\'][^"\']+["\']',
r'api_key\s*=\s*["\'][^"\']+["\']',
r'secret\s*=\s*["\'][^"\']+["\']'
]
findings = []
for root, dirs, files in os.walk('.'):
for file in files:
if file.endswith('.py'):
filepath = os.path.join(root, file)
with open(filepath, 'r') as f:
content = f.read()
for pattern in secret_patterns:
if re.search(pattern, content):
findings.append(filepath)
return findingsTesting von Automatisierungsskripten
1. Unit-Testing für Python-Skripte
Einfache Unit-Tests:
# test_backup.py
import unittest
import tempfile
import os
from backup_script import backup_files, validate_path
class TestBackupFunctions(unittest.TestCase):
def setUp(self):
"""Setup für Tests"""
self.temp_dir = tempfile.mkdtemp()
self.test_file = os.path.join(self.temp_dir, 'test.txt')
with open(self.test_file, 'w') as f:
f.write('Test content')
def tearDown(self):
"""Cleanup nach Tests"""
import shutil
shutil.rmtree(self.temp_dir)
def test_validate_path_existing(self):
"""Test für existierenden Pfad"""
self.assertTrue(validate_path(self.temp_dir))
def test_validate_path_non_existing(self):
"""Test für nicht-existierenden Pfad"""
self.assertFalse(validate_path('/nicht/existent'))
def test_backup_files_success(self):
"""Test für erfolgreiches Backup"""
backup_dir = os.path.join(self.temp_dir, 'backup')
os.makedirs(backup_dir)
result = backup_files([self.test_file], backup_dir)
self.assertTrue(result)
self.assertTrue(os.path.exists(os.path.join(backup_dir, 'test.txt')))
if __name__ == '__main__':
unittest.main()2. Mocking für externe Abhängigkeiten
Mocking von Dateisystem-Operationen:
from unittest.mock import patch, MagicMock
import unittest
class TestBackupWithMocking(unittest.TestCase):
@patch('backup_script.shutil.copy')
def test_backup_with_mock(self, mock_copy):
"""Test mit gemockter shutil.copy Funktion"""
mock_copy.return_value = True
result = backup_files(['file1.txt'], '/backup')
self.assertTrue(result)
mock_copy.assert_called_once()
@patch('backup_script.os.path.exists')
def test_validate_path_mock(self, mock_exists):
"""Test mit gemockter os.path.exists"""
mock_exists.return_value = True
result = validate_path('/any/path')
self.assertTrue(result)
mock_exists.assert_called_once_with('/any/path')3. Integration-Testing
End-to-End Tests:
# test_integration.py
import unittest
import tempfile
import os
import subprocess
class TestIntegration(unittest.TestCase):
def test_script_execution(self):
"""Test der kompletten Skript-Ausführung"""
script_path = 'backup.sh'
test_dir = tempfile.mkdtemp()
# Skript ausführen
result = subprocess.run(
['bash', script_path, test_dir],
capture_output=True,
text=True
)
# Ergebnis prüfen
self.assertEqual(result.returncode, 0)
self.assertIn('Backup erstellt', result.stdout)
# Cleanup
import shutil
shutil.rmtree(test_dir)4. Bash-Skripte testen
Bash-Unit-Testing Framework:
#!/bin/bash
# test_backup.sh - Tests für backup.sh
# Test-Funktionen
test_validate_path() {
source backup.sh
# Test mit existierendem Pfad
if validate_path "/tmp"; then
echo "✓ validate_path: existierender Pfad"
else
echo "✗ validate_path: existierender Pfad fehlgeschlagen"
exit 1
fi
# Test mit nicht-existierendem Pfad
if ! validate_path "/nicht/existent"; then
echo "✓ validate_path: nicht-existierender Pfad"
else
echo "✗ validate_path: nicht-existierender Pfad fehlgeschlagen"
exit 1
fi
}
test_backup_function() {
source backup.sh
local test_dir=$(mktemp -d)
local test_file="$test_dir/test.txt"
echo "Test content" > "$test_file"
# Backup-Funktion testen
backup_files "$test_file" "$test_dir/backup"
if [ -f "$test_dir/backup/test.txt" ]; then
echo "✓ backup_files: Datei erfolgreich kopiert"
else
echo "✗ backup_files: Datei nicht kopiert"
exit 1
fi
# Cleanup
rm -rf "$test_dir"
}
# Alle Tests ausführen
echo "=== Bash-Tests ==="
test_validate_path
test_backup_function
echo "Alle Tests bestanden!"5. PowerShell-Testing
Pester Framework für PowerShell:
# test_backup.ps1 - Pester Tests für backup.ps1
Describe "Backup-Funktionen" {
BeforeAll {
. .\backup.ps1
$testDir = Join-Path $env:TEMP "backup_test"
New-Item -Path $testDir -ItemType Directory -Force | Out-Null
}
AfterAll {
Remove-Item -Path $testDir -Recurse -Force -ErrorAction SilentlyContinue
}
Context "Validate-Path" {
It "sollte existierenden Pfad zurückgeben" {
$result = Validate-Path $testDir
$result | Should -Be $true
}
It "sollte nicht-existierenden Pfad ablehnen" {
$result = Validate-Path "C:\nicht\existent"
$result | Should -Be $false
}
}
Context "Backup-Files" {
It "sollte Dateien erfolgreich kopieren" {
$testFile = Join-Path $testDir "test.txt"
"Test content" | Out-File -FilePath $testFile
$backupDir = Join-Path $testDir "backup"
$result = Backup-Files $testFile $backupDir
$result | Should -Be $true
(Test-Path (Join-Path $backupDir "test.txt")) | Should -Be $true
}
}
}6. Test-Driven Development (TDD)
TDD-Workflow:
# 1. Test schreiben (der fehlschlägt)
def test_calculate_backup_size():
"""Test für Backup-Größenberechnung"""
result = calculate_backup_size('/tmp')
assert result > 0 # Wird fehlschlagen, Funktion existiert noch nicht
# 2. Funktion implementieren (damit Test bestanden wird)
def calculate_backup_size(directory: str) -> int:
"""Berechnet Größe eines Verzeichnisses"""
total_size = 0
for dirpath, dirnames, filenames in os.walk(directory):
for filename in filenames:
filepath = os.path.join(dirpath, filename)
total_size += os.path.getsize(filepath)
return total_size
# 3. Test ausführen (sollte jetzt bestanden werden)
# 4. Refactoring (während Tests weiterhin bestanden)7. Continuous Testing
Automatisierte Tests bei Änderungen:
# watchmedo für automatische Tests
pip install watchdog watchmedo
# Tests bei Dateiänderungen automatisch ausführen
watchmedo shell-command \
--patterns="*.py" \
--recursive \
--command='python -m pytest test_*.py -v'8. Test-Coverage
Coverage-Analyse:
# Python Coverage installieren
pip install pytest-cov
# Tests mit Coverage ausführen
pytest --cov=backup_script --cov-report=html
# Coverage-Report öffnen
open htmlcov/index.html9. Property-Based Testing
Hypothesis für Python:
from hypothesis import given, strategies as st
import unittest
class TestBackupProperties(unittest.TestCase):
@given(st.lists(st.text()))
def test_backup_preserves_content(self, content_list):
"""Property-Test: Backup erhält Inhalt bei"""
# Test mit verschiedenen Eingaben
for content in content_list:
with tempfile.NamedTemporaryFile(mode='w', delete=False) as f:
f.write(content)
temp_file = f.name
# Backup erstellen
backup_file = temp_file + '.backup'
shutil.copy(temp_file, backup_file)
# Inhalt prüfen
with open(backup_file, 'r') as f:
backup_content = f.read()
self.assertEqual(content, backup_content)
# Cleanup
os.unlink(temp_file)
os.unlink(backup_file)10. Load-Testing für Skripte
Performance-Tests:
import time
import statistics
def test_backup_performance():
"""Test der Backup-Performance"""
test_times = []
for i in range(10):
start_time = time.time()
backup_files(['large_file.txt'], '/backup')
end_time = time.time()
test_times.append(end_time - start_time)
avg_time = statistics.mean(test_times)
max_time = max(test_times)
print(f"Durchschnittszeit: {avg_time:.2f}s")
print(f"Maximalzeit: {max_time:.2f}s")
# Performance-Assertion
assert avg_time < 5.0, "Backup zu langsam"
assert max_time < 10.0, "Backup maximalzeit überschritten"Versionskontrolle für Automatisierungsskripte
1. Git-Grundlagen für Skripte
Initialisierung und erstes Commit:
# Git-Repository initialisieren
git init
# .gitignore erstellen
cat > .gitignore << EOF
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
env/
venv/
.venv/
# Secrets
.env
*.key
*.pem
secrets/
# Logs
*.log
# IDE
.vscode/
.idea/
*.swp
*.swo
# OS
.DS_Store
Thumbs.db
EOF
# Skripte hinzufügen
git add backup.sh backup.py config.yaml
git commit -m "Initial commit: Backup-Skripte hinzugefügt"2. Branching-Strategien
Git-Flow für Skripte:
# Main-Branch (produktiv)
git checkout -b main
# Develop-Branch (Entwicklung)
git checkout -b develop
# Feature-Branch für neue Funktion
git checkout -b feature/add-logging
# Änderungen commiten
git add backup.py
git commit -m "Add logging functionality"
# Zurück zu develop mergen
git checkout develop
git merge feature/add-logging
# Feature-Branch löschen
git branch -d feature/add-logging3. Commit-Messages konventionell
Conventional Commits:
# Format: <type>(<scope>): <description>
# Beispiele:
git commit -m "feat(backup): Add compression support"
git commit -m "fix(logging): Fix timestamp format"
git commit -m "docs(readme): Update installation instructions"
git commit -m "style(python): Apply PEP 8 formatting"
git commit -m "refactor(cleanup): Simplify error handling"
git commit -m "test(backup): Add unit tests for backup function"
git commit -m "chore(deps): Update dependencies"4. Code Reviews und Pull Requests
Pull-Request-Workflow:
# Feature-Branch erstellen
git checkout -b feature/backup-rotation
# Änderungen vornehmen und commiten
git add backup.sh
git commit -m "feat(backup): Add automatic backup rotation"
# Push zu Remote
git push origin feature/backup-rotation
# Pull Request erstellen (GitHub/GitLab/Bitbucket)
# Team-Mitglieder reviewen den Code
# Nach Approval mergen5. Git-Hooks für Qualitätssicherung
Pre-commit Hook:
#!/bin/bash
# .git/hooks/pre-commit
# Python-Code formatieren
autopep8 --in-place --aggressive *.py
# Python-Code linten
pylint *.py
# Tests ausführen
python -m pytest test_*.py
# Wenn alles erfolgreich, exit 0
exit 0Pre-push Hook:
#!/bin/bash
# .git/hooks/pre-push
# Alle Tests ausführen
python -m pytest
# Security-Check
safety check
# Wenn alles erfolgreich, exit 0
exit 06. Versionierung mit Tags
Semantic Versioning:
# Major-Version (inkompatible Änderungen)
git tag -a v2.0.0 -m "Major release: Breaking changes"
git push origin v2.0.0
# Minor-Version (neue Features, rückwärtskompatibel)
git tag -a v1.2.0 -m "Minor release: New features"
git push origin v1.2.0
# Patch-Version (Bugfixes, rückwärtskompatibel)
git tag -a v1.1.1 -m "Patch release: Bug fixes"
git push origin v1.1.17. Changelog führen
Automatisches Changelog:
# Mit conventional-changelog
npm install -g conventional-changelog-cli
# Changelog generieren
conventional-changelog -p angular -i CHANGELOG.md -s
# CHANGELOG.md Struktur:
# ## [1.2.0] - 2026-07-14
# ### Added
# - Backup-Rotation hinzugefügt
# - Logging-Funktionalität implementiert
#
# ### Fixed
# - Fehler bei Pfad-Validierung behoben
#
# ## [1.1.0] - 2026-06-01
# ### Added
# - Erste Version des Backup-Skripts8. Konfiguration in Git
Git-Konfiguration für Skripte:
# Benutzer konfigurieren
git config user.name "Dein Name"
git config user.email "deine@email.com"
# Line-Endings (wichtig für Cross-Platform)
git config core.autocrlf input # Linux/Mac
git config core.autocrlf true # Windows
# Datei-Attribute
git config core.filemode false # Permissions ignorieren
# Diff-Tool konfigurieren
git config diff.tool vimdiff
git config merge.tool vimdiff9. Git-Workflows für Teams
Forking-Workflow:
# 1. Repository forken (GitHub/GitLab)
# 2. Fork klonen
git clone https://github.com/dein-username/repo.git
# 3. Upstream remote hinzufügen
git remote add upstream https://github.com/original/repo.git
# 4. Feature-Branch erstellen
git checkout -b feature/neue-funktion
# 5. Änderungen commiten
git add script.py
git commit -m "feat: neue Funktion hinzugefügt"
# 6. Push zu deinem Fork
git push origin feature/neue-funktion
# 7. Pull Request erstellen
# 8. Nach Review mergen
# 9. Upstream changes holen
git fetch upstream
git checkout main
git merge upstream/main10. Git für Backup-Strategien
Git als Backup-System:
#!/bin/bash
# git_backup.sh - Automatisches Git-Backup
REPO_DIR="/path/to/repo"
BACKUP_REMOTE="git@github.com:user/backup-repo.git"
cd "$REPO_DIR"
# Änderungen hinzufügen
git add .
# Commit mit Timestamp
git commit -m "Auto-backup: $(date '+%Y-%m-%d %H:%M:%S')"
# Push zu Remote
git push "$BACKUP_REMOTE" main
echo "Backup erfolgreich nach Git gepusht"CI/CD Integration für Automatisierungsskripte
1. GitHub Actions für Python-Skripte
Workflow-Datei:
# .github/workflows/automation.yml
name: Automation Scripts CI/CD
on:
push:
branches: [ main, develop ]
pull_request:
branches: [ main ]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install pytest pytest-cov pylint safety
- name: Lint with pylint
run: pylint *.py
- name: Security check
run: safety check
- name: Run tests
run: pytest --cov=. --cov-report=xml
- name: Upload coverage
uses: codecov/codecov-action@v3
with:
file: ./coverage.xml
deploy:
needs: test
runs-on: ubuntu-latest
if: github.ref == 'refs/heads/main'
steps:
- uses: actions/checkout@v3
- name: Deploy to production
run: |
echo "Deploying automation scripts to production"
# Deployment-Logik hier2. GitLab CI für Bash-Skripte
GitLab CI-Datei:
# .gitlab-ci.yml
stages:
- test
- deploy
variables:
SCRIPT_DIR: "/opt/automation"
test_scripts:
stage: test
image: alpine:latest
before_script:
- apk add --no-cache bash shellcheck
script:
- shellcheck *.sh
- bash -n backup.sh
- bash test_backup.sh
artifacts:
reports:
junit: test-results.xml
deploy_production:
stage: deploy
image: alpine:latest
only:
- main
script:
- apk add --no-cache rsync openssh
- mkdir -p ~/.ssh
- echo "$SSH_PRIVATE_KEY" > ~/.ssh/id_rsa
- chmod 600 ~/.ssh/id_rsa
- rsync -avz --delete *.sh user@server:$SCRIPT_DIR/
- ssh user@server "chmod +x $SCRIPT_DIR/*.sh"3. Jenkins Pipeline
Jenkinsfile:
// Jenkinsfile
pipeline {
agent any
environment {
PYTHON_VERSION = '3.11'
DEPLOY_SERVER = 'user@server'
}
stages {
stage('Checkout') {
steps {
checkout scm
}
}
stage('Lint') {
steps {
sh 'pylint *.py || true'
sh 'shellcheck *.sh || true'
}
}
stage('Test') {
steps {
sh 'python -m pytest test_*.py --cov=. --cov-report=html'
}
}
stage('Security Scan') {
steps {
sh 'safety check'
}
}
stage('Deploy') {
when {
branch 'main'
}
steps {
sh '''
rsync -avz --delete *.sh ${DEPLOY_SERVER}:/opt/automation/
ssh ${DEPLOY_SERVER} "chmod +x /opt/automation/*.sh"
'''
}
}
}
post {
always {
archiveArtifacts artifacts: 'htmlcov/**', allowEmptyArchive: true
publishHTML target: [
reportDir: 'htmlcov',
reportFiles: 'index.html',
reportName: 'Coverage Report'
]
}
}
}4. Azure DevOps Pipeline
azure-pipelines.yml:
# azure-pipelines.yml
trigger:
branches:
include:
- main
- develop
pool:
vmImage: 'ubuntu-latest'
variables:
pythonVersion: '3.11'
stages:
- stage: Test
jobs:
- job: TestScripts
steps:
- task: UsePythonVersion@0
inputs:
versionSpec: '$(pythonVersion)'
- script: |
python -m pip install --upgrade pip
pip install pytest pytest-cov pylint safety
displayName: 'Install dependencies'
- script: |
pylint *.py
displayName: 'Lint Python'
- script: |
safety check
displayName: 'Security check'
- script: |
pytest --cov=. --cov-report=xml
displayName: 'Run tests'
- task: PublishCodeCoverageResults@1
inputs:
codeCoverageTool: 'Cobertura'
summaryFileLocation: '$(System.DefaultWorkingDirectory)/coverage.xml'
- stage: Deploy
dependsOn: Test
condition: and(succeeded(), eq(variables['Build.SourceBranch'], 'refs/heads/main'))
jobs:
- deployment: DeployScripts
environment: 'Production'
strategy:
runOnce:
deploy:
steps:
- script: |
rsync -avz --delete *.sh user@server:/opt/automation/
displayName: 'Deploy scripts'5. Docker-basierte CI/CD
Dockerfile für Tests:
# Dockerfile.test
FROM python:3.11-slim
# System-Abhängigkeiten
RUN apt-get update && apt-get install -y \
shellcheck \
bash \
&& rm -rf /var/lib/apt/lists/*
# Python-Abhängigkeiten
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN pip install --no-cache-dir pytest pytest-cov pylint safety
# Skripte kopieren
COPY *.py *.sh /app/
WORKDIR /app
# Test-Command
CMD ["pytest", "--cov=.", "--cov-report=html"]Docker Compose für CI:
# docker-compose.test.yml
version: '3.8'
services:
test:
build:
context: .
dockerfile: Dockerfile.test
volumes:
- .:/app
- ./htmlcov:/app/htmlcov
command: pytest --cov=. --cov-report=html6. Environment Management
Multi-Environment Deployment:
# .github/workflows/deploy.yml
name: Multi-Environment Deploy
on:
push:
branches: [ main ]
jobs:
deploy_staging:
runs-on: ubuntu-latest
environment: staging
steps:
- uses: actions/checkout@v3
- name: Deploy to Staging
run: |
rsync -avz --delete *.sh $STAGING_SERVER:/opt/automation/
env:
STAGING_SERVER: ${{ secrets.STAGING_SERVER }}
deploy_production:
runs-on: ubuntu-latest
environment: production
needs: deploy_staging
steps:
- uses: actions/checkout@v3
- name: Deploy to Production
run: |
rsync -avz --delete *.sh $PROD_SERVER:/opt/automation/
env:
PROD_SERVER: ${{ secrets.PROD_SERVER }}7. Rollback-Strategien
Automatisches Rollback:
# .github/workflows/rollback.yml
name: Rollback Deployment
on:
workflow_dispatch:
inputs:
version:
description: 'Version to rollback to'
required: true
jobs:
rollback:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
with:
ref: ${{ github.event.inputs.version }}
- name: Deploy previous version
run: |
rsync -avz --delete *.sh $PROD_SERVER:/opt/automation/
env:
PROD_SERVER: ${{ secrets.PROD_SERVER }}
- name: Notify team
run: |
echo "Rollback to version ${{ github.event.inputs.version }} completed"8. Monitoring in CI/CD
Pipeline-Health-Checks:
# .github/workflows/monitoring.yml
name: Pipeline Monitoring
on:
schedule:
- cron: '0 * * * *' # Jede Stunde
jobs:
health_check:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Check script health
run: |
# Prüfen ob Skripte laufen
ssh $PROD_SERVER "ps aux | grep backup.sh"
# Prüfen ob Logs geschrieben werden
ssh $PROD_SERVER "tail -n 10 /var/log/automation.log"
env:
PROD_SERVER: ${{ secrets.PROD_SERVER }}
- name: Alert on failure
if: failure()
run: |
# Alert senden (Slack, Email, etc.)
curl -X POST $SLACK_WEBHOOK -d '{"text":"Automation health check failed"}'
env:
SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK }}9. Automated Testing in Pipeline
Test-Matrix für verschiedene Python-Versionen:
# .github/workflows/test-matrix.yml
name: Test Matrix
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ['3.9', '3.10', '3.11']
steps:
- uses: actions/checkout@v3
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v4
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install pytest
- name: Run tests
run: pytest test_*.py10. Secrets Management in CI/CD
GitHub Secrets verwenden:
# .github/workflows/secure-deploy.yml
name: Secure Deployment
on:
push:
branches: [ main ]
jobs:
deploy:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Setup SSH key
run: |
mkdir -p ~/.ssh
echo "${{ secrets.SSH_PRIVATE_KEY }}" > ~/.ssh/id_rsa
chmod 600 ~/.ssh/id_rsa
ssh-keyscan -H ${{ secrets.DEPLOY_SERVER }} >> ~/.ssh/known_hosts
- name: Deploy with secrets
run: |
# Secrets aus Environment laden
export DB_PASSWORD="${{ secrets.DB_PASSWORD }}"
export API_KEY="${{ secrets.API_KEY }}"
# Skript ausführen
rsync -avz --delete *.sh ${{ secrets.DEPLOY_SERVER }}:/opt/automation/
ssh ${{ secrets.DEPLOY_SERVER }} "cd /opt/automation && ./deploy.sh"
env:
DEPLOY_SERVER: ${{ secrets.DEPLOY_SERVER }}Containerisierung von Automatisierungsskripten
1. Dockerfile für Python-Skripte
Einfaches Dockerfile:
# Dockerfile
FROM python:3.11-slim
# Arbeitsverzeichnis
WORKDIR /app
# Abhängigkeiten kopieren
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Skripte kopieren
COPY backup.py .
COPY config.yaml .
# Skript als Entry-Point
ENTRYPOINT ["python", "backup.py"]
CMD ["--config", "config.yaml"]2. Dockerfile für Bash-Skripte
Bash im Container:
# Dockerfile.bash
FROM alpine:latest
# Bash und Tools installieren
RUN apk add --no-cache bash rsync coreutils
# Skripte kopieren
COPY backup.sh /usr/local/bin/
RUN chmod +x /usr/local/bin/backup.sh
# Konfiguration kopieren
COPY config.yaml /etc/backup/
# Working Directory
WORKDIR /data
ENTRYPOINT ["/usr/local/bin/backup.sh"]3. Multi-Stage Dockerfile
Optimiertes Multi-Stage Build:
# Dockerfile.multi-stage
# Build Stage
FROM python:3.11-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --user --no-cache-dir -r requirements.txt
# Runtime Stage
FROM python:3.11-slim
WORKDIR /app
COPY --from=builder /root/.local /root/.local
COPY backup.py .
COPY config.yaml .
ENV PATH=/root/.local/bin:$PATH
ENTRYPOINT ["python", "backup.py"]4. Docker Compose für komplette Umgebung
docker-compose.yml:
version: '3.8'
services:
backup:
build:
context: .
dockerfile: Dockerfile
volumes:
- ./data:/data
- ./config:/config
- ./logs:/logs
environment:
- BACKUP_DIR=/data/backups
- SOURCE_DIR=/data/source
- LOG_LEVEL=INFO
restart: unless-stopped
monitoring:
image: prom/prometheus
ports:
- "9090:9090"
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
grafana:
image: grafana/grafana
ports:
- "3000:3000"
volumes:
- grafana-data:/var/lib/grafana
volumes:
grafana-data:5. Kubernetes Deployment
Deployment YAML:
# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: automation-scripts
spec:
replicas: 2
selector:
matchLabels:
app: automation
template:
metadata:
labels:
app: automation
spec:
containers:
- name: backup
image: your-registry/backup:latest
env:
- name: BACKUP_DIR
value: "/data/backups"
- name: SOURCE_DIR
valueFrom:
configMapKeyRef:
name: backup-config
key: source_dir
volumeMounts:
- name: data
mountPath: /data
- name: config
mountPath: /config
volumes:
- name: data
persistentVolumeClaim:
claimName: backup-pvc
- name: config
configMap:
name: backup-config6. Docker Health Checks
Health Check im Dockerfile:
# Dockerfile mit Health Check
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY backup.py health_check.py .
# Health Check Script
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD python health_check.py || exit 1
ENTRYPOINT ["python", "backup.py"]Health Check Script:
# health_check.py
import os
import sys
def check_health():
"""Prüft ob Backup-Skript gesund ist"""
# Prüfen ob Konfiguration existiert
if not os.path.exists('config.yaml'):
print("Config file missing")
return False
# Prüfen ob Backup-Verzeichnis beschreibbar ist
backup_dir = os.getenv('BACKUP_DIR', '/data/backups')
if not os.access(backup_dir, os.W_OK):
print("Backup directory not writable")
return False
print("Health check passed")
return True
if __name__ == '__main__':
sys.exit(0 if check_health() else 1)7. Container-Optimierung
Best Practices für kleine Images:
# Optimiertes Dockerfile
FROM python:3.11-alpine AS builder
# Nur notwendige Abhängigkeiten
COPY requirements.txt .
RUN pip install --no-cache-dir --user --no-deps -r requirements.txt
FROM python:3.11-alpine
# Nicht-root User
RUN addgroup -g 1000 backup && \
adduser -D -u 1000 -G backup backup
WORKDIR /app
COPY --from=builder /root/.local /root/.local
COPY --chown=backup:backup backup.py .
USER backup
ENV PATH=/root/.local/bin:$PATH
ENTRYPOINT ["python", "backup.py"]8. Container-Security
Security-Scan und Hardening:
# Security-hardened Dockerfile
FROM python:3.11-slim
# Security Updates
RUN apt-get update && \
apt-get upgrade -y && \
apt-get install -y --no-install-recommends \
ca-certificates && \
rm -rf /var/lib/apt/lists/*
# Nicht-root User
RUN groupadd -r backup && useradd -r -g backup backup
WORKDIR /app
COPY --chown=backup:backup requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt
COPY --chown=backup:backup backup.py .
USER backup
ENTRYPOINT ["python", "backup.py"]Security-Scan mit Trivy:
# Image scannen
trivy image your-registry/backup:latest
# In CI/CD integrieren
trivy image --exit-code 1 --severity HIGH,CRITICAL your-registry/backup:latest9. Container-Logging
Strukturiertes Logging:
# backup.py mit Container-Logging
import logging
import json
import sys
class JSONFormatter(logging.Formatter):
"""JSON Formatter für Container-Logs"""
def format(self, record):
log_data = {
'timestamp': self.formatTime(record),
'level': record.levelname,
'message': record.getMessage(),
'logger': record.name
}
if hasattr(record, 'extra_data'):
log_data.update(record.extra_data)
return json.dumps(log_data)
# Logging konfigurieren
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
logger = logging.getLogger(__name__)
logger.addHandler(handler)
logger.setLevel(logging.INFO)10. Container-Orchestrierung
Docker Swarm Service:
# docker-stack.yml
version: '3.8'
services:
backup:
image: your-registry/backup:latest
deploy:
replicas: 3
update_config:
parallelism: 1
delay: 10s
restart_policy:
condition: on-failure
delay: 5s
max_attempts: 3
environment:
- BACKUP_DIR=/data/backups
volumes:
- backup-data:/data
networks:
- automation-network
networks:
automation-network:
driver: overlay
volumes:
backup-data:Monitoring & Alerting für Automatisierungsskripte
1. Prometheus-Metrik-Export
Python-Metriken mit Prometheus:
# metrics_exporter.py
from prometheus_client import start_http_server, Counter, Histogram, Gauge
import time
# Metriken definieren
backup_counter = Counter('backup_total', 'Total number of backups', ['status'])
backup_duration = Histogram('backup_duration_seconds', 'Backup duration')
backup_size = Gauge('backup_size_bytes', 'Backup size in bytes')
def backup_with_metrics(source: str, target: str):
"""Backup mit Prometheus-Metriken"""
start_time = time.time()
try:
# Backup durchführen
size = perform_backup(source, target)
# Metriken aktualisieren
backup_counter.labels(status='success').inc()
backup_duration.observe(time.time() - start_time)
backup_size.set(size)
except Exception as e:
backup_counter.labels(status='failed').inc()
raise
# Metrics-Server starten
if __name__ == '__main__':
start_http_server(8000)
print("Metrics server started on port 8000")2. Grafana-Dashboard
Grafana Dashboard-Konfiguration:
{
"dashboard": {
"title": "Automation Scripts Monitoring",
"panels": [
{
"title": "Backup Success Rate",
"targets": [
{
"expr": "rate(backup_total{status='success'}[5m])"
}
]
},
{
"title": "Backup Duration",
"targets": [
{
"expr": "histogram_quantile(0.95, rate(backup_duration_seconds_bucket[5m]))"
}
]
},
{
"title": "Backup Size",
"targets": [
{
"expr": "backup_size_bytes"
}
]
}
]
}
}3. Log-Analyse mit ELK Stack
Filebeat-Konfiguration:
# filebeat.yml
filebeat.inputs:
- type: log
enabled: true
paths:
- /var/log/automation/*.log
fields:
app: automation
fields_under_root: true
output.elasticsearch:
hosts: ["localhost:9200"]
index: "automation-logs-%{+yyyy.MM.dd}"
setup.kibana:
host: "localhost:5601"4. Alerting mit Alertmanager
Alertmanager-Konfiguration:
# alertmanager.yml
global:
resolve_timeout: 5m
route:
group_by: ['alertname']
group_wait: 10s
group_interval: 10s
repeat_interval: 1h
receiver: 'web.hook'
receivers:
- name: 'web.hook'
webhook_configs:
- url: 'http://localhost:5001/alerts'
- name: 'slack'
slack_configs:
- api_url: 'YOUR_SLACK_WEBHOOK_URL'
channel: '#automation-alerts'Prometheus-Alert-Regeln:
# alerts.yml
groups:
- name: automation_alerts
rules:
- alert: BackupFailed
expr: rate(backup_total{status='failed'}[5m]) > 0
for: 1m
labels:
severity: critical
annotations:
summary: "Backup failed"
description: "Backup has been failing for the last 5 minutes"
- alert: BackupTooSlow
expr: histogram_quantile(0.95, rate(backup_duration_seconds_bucket[5m])) > 300
for: 5m
labels:
severity: warning
annotations:
summary: "Backup too slow"
description: "Backup duration exceeds 5 minutes"5. Health-Check-Endpoints
HTTP-Health-Check:
# health_check.py
from flask import Flask, jsonify
import os
import psutil
app = Flask(__name__)
@app.route('/health')
def health_check():
"""Health-Check Endpoint"""
health_status = {
'status': 'healthy',
'checks': {}
}
# Disk-Space prüfen
disk_usage = psutil.disk_usage('/')
health_status['checks']['disk'] = {
'status': 'ok' if disk_usage.percent < 90 else 'warning',
'usage_percent': disk_usage.percent
}
# Memory prüfen
memory = psutil.virtual_memory()
health_status['checks']['memory'] = {
'status': 'ok' if memory.percent < 90 else 'warning',
'usage_percent': memory.percent
}
# Config-File prüfen
config_exists = os.path.exists('config.yaml')
health_status['checks']['config'] = {
'status': 'ok' if config_exists else 'error',
'exists': config_exists
}
# Overall Status
if any(check['status'] == 'error' for check in health_status['checks'].values()):
health_status['status'] = 'error'
elif any(check['status'] == 'warning' for check in health_status['checks'].values()):
health_status['status'] = 'warning'
return jsonify(health_status)
if __name__ == '__main__':
app.run(host='0.0.0.0', port=8080)6. Slack-Integration
Slack-Alerts senden:
# slack_alerts.py
import requests
import json
def send_slack_alert(webhook_url: str, message: str, level: str = 'info'):
"""Sendet Alert an Slack"""
colors = {
'info': '#36a64f',
'warning': '#ff9900',
'error': '#ff0000'
}
payload = {
'attachments': [{
'color': colors.get(level, '#36a64f'),
'title': f'Automation Alert: {level.upper()}',
'text': message,
'footer': 'Automation Scripts',
'ts': int(time.time())
}]
}
response = requests.post(webhook_url, json=payload)
return response.status_code == 200
# Nutzung
send_slack_alert(
webhook_url='YOUR_SLACK_WEBHOOK',
message='Backup failed for database',
level='error'
)7. Email-Alerting
Email-Alerts mit Python:
# email_alerts.py
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
def send_email_alert(
subject: str,
message: str,
to_email: str,
from_email: str,
smtp_server: str,
smtp_port: int = 587
):
"""Sendet Email-Alert"""
msg = MIMEMultipart()
msg['From'] = from_email
msg['To'] = to_email
msg['Subject'] = subject
msg.attach(MIMEText(message, 'plain'))
with smtplib.SMTP(smtp_server, smtp_port) as server:
server.starttls()
server.login(from_email, 'YOUR_PASSWORD')
server.send_message(msg)
# Nutzung
send_email_alert(
subject='Automation Alert: Backup Failed',
message='The backup script failed with error: ...',
to_email='admin@example.com',
from_email='automation@example.com',
smtp_server='smtp.example.com'
)8. System-Monitoring mit psutil
System-Ressourcen überwachen:
# system_monitor.py
import psutil
import time
def monitor_system(interval: int = 60):
"""Überwacht System-Ressourcen"""
while True:
# CPU
cpu_percent = psutil.cpu_percent(interval=1)
# Memory
memory = psutil.virtual_memory()
# Disk
disk = psutil.disk_usage('/')
# Network
network = psutil.net_io_counters()
# Ausgabe
print(f"CPU: {cpu_percent}%")
print(f"Memory: {memory.percent}%")
print(f"Disk: {disk.percent}%")
print(f"Network Sent: {network.bytes_sent} bytes")
print(f"Network Recv: {network.bytes_recv} bytes")
print("-" * 40)
# Alert bei hohen Werten
if cpu_percent > 90:
send_alert(f"High CPU usage: {cpu_percent}%")
if memory.percent > 90:
send_alert(f"High memory usage: {memory.percent}%")
if disk.percent > 90:
send_alert(f"High disk usage: {disk.percent}%")
time.sleep(interval)9. Log-Rotation und Management
Logrotate-Konfiguration:
# /etc/logrotate.d/automation
/var/log/automation/*.log {
daily
rotate 14
compress
delaycompress
missingok
notifempty
create 0640 automation automation
sharedscripts
postrotate
systemctl reload automation-service > /dev/null 2>&1 || true
endscript
}10. Uptime-Monitoring
Uptime-Check mit Python:
# uptime_monitor.py
import requests
import time
from datetime import datetime
def check_uptime(url: str, interval: int = 60):
"""Überwacht Uptime eines Services"""
downtime_start = None
total_downtime = 0
while True:
try:
response = requests.get(url, timeout=10)
if response.status_code == 200:
if downtime_start:
downtime_duration = time.time() - downtime_start
total_downtime += downtime_duration
send_alert(f"Service is back up after {downtime_duration:.2f}s downtime")
downtime_start = None
print(f"{datetime.now()}: Service is UP")
else:
if not downtime_start:
downtime_start = time.time()
send_alert(f"Service returned status {response.status_code}")
print(f"{datetime.now()}: Service returned {response.status_code}")
except requests.RequestException as e:
if not downtime_start:
downtime_start = time.time()
send_alert(f"Service is down: {str(e)}")
print(f"{datetime.now()}: Service is DOWN - {str(e)}")
time.sleep(interval)Erweiterte Fehlerbehandlung für Automatisierungsskripte
1. Strukturierte Fehlerbehandlung
Custom Exception Classes:
# exceptions.py
class AutomationError(Exception):
"""Base Exception für Automation-Skripte"""
pass
class BackupError(AutomationError):
"""Backup-spezifische Fehler"""
pass
class ConfigError(AutomationError):
"""Konfigurations-Fehler"""
pass
class NetworkError(AutomationError):
"""Netzwerk-Fehler"""
pass
class ValidationError(AutomationError):
"""Validierungs-Fehler"""
pass2. Retry-Mechanismen mit Exponential Backoff
Intelligentes Retry:
# retry_handler.py
import time
import random
from functools import wraps
from typing import Callable, Type
def retry_with_backoff(
max_retries: int = 3,
initial_delay: float = 1.0,
max_delay: float = 60.0,
exponential_base: float = 2,
jitter: bool = True,
exceptions: tuple = (Exception,)
):
"""Decorator mit Exponential Backoff und Jitter"""
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
delay = initial_delay
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except exceptions as e:
if attempt == max_retries - 1:
raise
# Jitter hinzufügen um Thundering Herd zu vermeiden
if jitter:
delay = delay * (1 + random.uniform(-0.1, 0.1))
print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay:.2f}s...")
time.sleep(min(delay, max_delay))
delay *= exponential_base
return wrapper
return decorator
# Nutzung
@retry_with_backoff(max_retries=5, initial_delay=2.0, exceptions=(ConnectionError, TimeoutError))
def download_file(url: str, target: str):
"""Lädt Datei mit Retry herunter"""
# Download-Logik
pass3. Circuit Breaker Pattern
Circuit Breaker für externe Services:
# circuit_breaker.py
from enum import Enum
import time
class CircuitState(Enum):
CLOSED = "closed" # Normal operation
OPEN = "open" # Circuit is open, requests fail immediately
HALF_OPEN = "half_open" # Testing if service recovered
class CircuitBreaker:
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: float = 60.0,
expected_exception: Exception = Exception
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.expected_exception = expected_exception
self.failure_count = 0
self.state = CircuitState.CLOSED
self.last_failure_time = None
def call(self, func: Callable, *args, **kwargs):
"""Führt Funktion mit Circuit Breaker aus"""
if self.state == CircuitState.OPEN:
if self._should_attempt_reset():
self.state = CircuitState.HALF_OPEN
else:
raise Exception("Circuit breaker is OPEN")
try:
result = func(*args, **kwargs)
self._on_success()
return result
except self.expected_exception as e:
self._on_failure()
raise
def _should_attempt_reset(self) -> bool:
"""Prüft ob Circuit zurückgesetzt werden sollte"""
return (
self.last_failure_time and
time.time() - self.last_failure_time >= self.recovery_timeout
)
def _on_success(self):
"""Bei erfolgreicher Ausführung"""
self.failure_count = 0
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.CLOSED
def _on_failure(self):
"""Bei fehlgeschlagener Ausführung"""
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
# Nutzung
circuit_breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=30)
def call_external_api():
"""Externer API-Call mit Circuit Breaker"""
return circuit_breaker.call(requests.get, "https://api.example.com")4. Dead Letter Queue für fehlgeschlagene Aufgaben
Dead Letter Queue Implementierung:
# dead_letter_queue.py
import json
import os
from datetime import datetime
from typing import Any, Dict
class DeadLetterQueue:
def __init__(self, queue_file: str = "dead_letter_queue.json"):
self.queue_file = queue_file
self._load_queue()
def _load_queue(self):
"""Lädt Queue aus Datei"""
if os.path.exists(self.queue_file):
with open(self.queue_file, 'r') as f:
self.queue = json.load(f)
else:
self.queue = []
def _save_queue(self):
"""Speichert Queue in Datei"""
with open(self.queue_file, 'w') as f:
json.dump(self.queue, f, indent=2)
def add(self, task: Dict[str, Any], error: Exception):
"""Fügt fehlgeschlagene Aufgabe zur Queue hinzu"""
dead_letter = {
'task': task,
'error': str(error),
'error_type': type(error).__name__,
'timestamp': datetime.now().isoformat(),
'retry_count': 0
}
self.queue.append(dead_letter)
self._save_queue()
def get_next(self) -> Dict[str, Any]:
"""Holt nächste Aufgabe aus Queue"""
if not self.queue:
return None
return self.queue.pop(0)
def retry(self, max_retries: int = 3):
"""Versucht Aufgaben erneut"""
remaining_tasks = []
for dead_letter in self.queue:
if dead_letter['retry_count'] < max_retries:
try:
# Aufgabe erneut ausführen
execute_task(dead_letter['task'])
print(f"Task {dead_letter['task']['id']} erfolgreich retried")
except Exception as e:
dead_letter['retry_count'] += 1
dead_letter['last_error'] = str(e)
remaining_tasks.append(dead_letter)
else:
print(f"Task {dead_letter['task']['id']} hat max retries erreicht")
self.queue = remaining_tasks
self._save_queue()
# Nutzung
dlq = DeadLetterQueue()
try:
execute_task(task)
except Exception as e:
dlq.add(task, e)5. Graceful Degradation
Fallback-Strategien:
# graceful_degradation.py
from typing import Optional, Callable
class FallbackChain:
def __init__(self):
self.fallbacks = []
def add_fallback(self, func: Callable, priority: int = 0):
"""Fügt Fallback-Funktion hinzu"""
self.fallbacks.append((priority, func))
self.fallbacks.sort(key=lambda x: x[0], reverse=True)
def execute(self, *args, **kwargs) -> Optional[Any]:
"""Führt Funktionen mit Fallback aus"""
for priority, func in self.fallbacks:
try:
result = func(*args, **kwargs)
if result is not None:
return result
except Exception as e:
print(f"Fallback {func.__name__} failed: {e}")
continue
return None
# Nutzung
fallback_chain = FallbackChain()
# Primäre Methode
def primary_backup(source: str, target: str):
"""Primäre Backup-Methode (Cloud)"""
return upload_to_cloud(source, target)
# Fallback 1
def secondary_backup(source: str, target: str):
"""Sekundäre Backup-Methode (FTP)"""
return upload_to_ftp(source, target)
# Fallback 2
def tertiary_backup(source: str, target: str):
"""Tertiäre Backup-Methode (Lokal)"""
return copy_locally(source, target)
fallback_chain.add_fallback(primary_backup, priority=3)
fallback_chain.add_fallback(secondary_backup, priority=2)
fallback_chain.add_fallback(tertiary_backup, priority=1)
result = fallback_chain.execute(source, target)6. Comprehensive Logging für Fehler
Detailliertes Error-Logging:
# error_logger.py
import logging
import traceback
import sys
from datetime import datetime
from typing import Optional
class ErrorLogger:
def __init__(self, log_file: str = "automation_errors.log"):
self.logger = logging.getLogger('error_logger')
self.logger.setLevel(logging.ERROR)
# File Handler
file_handler = logging.FileHandler(log_file)
file_handler.setLevel(logging.ERROR)
# Formatter
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
file_handler.setFormatter(formatter)
self.logger.addHandler(file_handler)
def log_error(
self,
error: Exception,
context: Optional[dict] = None,
critical: bool = False
):
"""Loggt Fehler mit Kontext"""
error_info = {
'error_type': type(error).__name__,
'error_message': str(error),
'traceback': traceback.format_exc(),
'context': context or {},
'timestamp': datetime.now().isoformat()
}
log_message = f"Error: {error_info}"
if critical:
self.logger.critical(log_message)
else:
self.logger.error(log_message)
return error_info
def log_exception(self, func: Callable):
"""Decorator für Exception-Logging"""
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
self.log_error(e, context={'function': func.__name__})
raise
return wrapper
# Nutzung
error_logger = ErrorLogger()
@error_logger.log_exception
def risky_operation():
"""Risikoreiche Operation mit automatischem Logging"""
pass7. Error Recovery Strategien
Automatische Recovery:
# recovery.py
from typing import Callable, Any
class RecoveryManager:
def __init__(self):
self.recovery_strategies = {}
def register_recovery(self, error_type: type, recovery_func: Callable):
"""Registriert Recovery-Strategie für Fehler-Typ"""
self.recovery_strategies[error_type] = recovery_func
def execute_with_recovery(self, func: Callable, *args, **kwargs) -> Any:
"""Führt Funktion mit Recovery aus"""
try:
return func(*args, **kwargs)
except Exception as e:
error_type = type(e)
if error_type in self.recovery_strategies:
print(f"Attempting recovery for {error_type.__name__}")
recovery_func = self.recovery_strategies[error_type]
return recovery_func(e, *args, **kwargs)
else:
raise
# Nutzung
recovery_manager = RecoveryManager()
def recover_from_connection_error(error: Exception, *args, **kwargs):
"""Recovery von Verbindungsfehlern"""
print("Reconnecting...")
time.sleep(5)
return func(*args, **kwargs)
def recover_from_disk_full(error: Exception, *args, **kwargs):
"""Recovery von vollen Festplatten"""
print("Cleaning up old files...")
cleanup_old_files()
return func(*args, **kwargs)
recovery_manager.register_recovery(ConnectionError, recover_from_connection_error)
recovery_manager.register_recovery(OSError, recover_from_disk_full)8. Timeout-Handling
Timeout für langlaufende Operationen:
# timeout_handler.py
import signal
from contextlib import contextmanager
import time
class TimeoutError(Exception):
"""Custom Timeout Exception"""
pass
@contextmanager
def timeout_handler(seconds: int):
"""Context Manager für Timeout"""
def signal_handler(signum, frame):
raise TimeoutError(f"Operation timed out after {seconds} seconds")
# Signal Handler registrieren
old_handler = signal.signal(signal.SIGALRM, signal_handler)
signal.alarm(seconds)
try:
yield
finally:
# Signal Handler zurücksetzen
signal.alarm(0)
signal.signal(signal.SIGALRM, old_handler)
# Nutzung
try:
with timeout_handler(30):
long_running_operation()
except TimeoutError:
print("Operation timed out, using fallback")
fallback_operation()9. Error Aggregation
Mehrere Fehler sammeln:
# error_aggregator.py
from typing import List, Dict, Any
class ErrorAggregator:
def __init__(self):
self.errors: List[Dict[str, Any]] = []
def add_error(self, error: Exception, context: Dict[str, Any] = None):
"""Fügt Fehler zur Sammlung hinzu"""
self.errors.append({
'error': str(error),
'type': type(error).__name__,
'context': context or {},
'timestamp': time.time()
})
def has_errors(self) -> bool:
"""Prüft ob Fehler vorhanden sind"""
return len(self.errors) > 0
def get_summary(self) -> Dict[str, Any]:
"""Gibt Zusammenfassung der Fehler zurück"""
error_types = {}
for error in self.errors:
error_type = error['type']
error_types[error_type] = error_types.get(error_type, 0) + 1
return {
'total_errors': len(self.errors),
'error_types': error_types,
'errors': self.errors
}
def clear(self):
"""Löscht alle Fehler"""
self.errors = []
# Nutzung
aggregator = ErrorAggregator()
for item in items:
try:
process_item(item)
except Exception as e:
aggregator.add_error(e, context={'item': item})
if aggregator.has_errors():
print(f"Processing completed with {len(aggregator.errors)} errors")
print(aggregator.get_summary())10. Error Notification System
Benachrichtigung bei kritischen Fehlern:
# error_notifier.py
from typing import List, Callable
class ErrorNotifier:
def __init__(self):
self.notifiers: List[Callable] = []
def add_notifier(self, notifier: Callable):
"""Fügt Benachrichtigungs-Funktion hinzu"""
self.notifiers.append(notifier)
def notify(self, error: Exception, context: dict = None):
"""Sendet Benachrichtigung an alle Notifier"""
error_info = {
'error': str(error),
'type': type(error).__name__,
'context': context or {},
'timestamp': time.time()
}
for notifier in self.notifiers:
try:
notifier(error_info)
except Exception as e:
print(f"Notifier failed: {e}")
# Beispiel Notifier
def slack_notifier(error_info: dict):
"""Slack Benachrichtigung"""
send_slack_message(f"Error: {error_info['error']}")
def email_notifier(error_info: dict):
"""Email Benachrichtigung"""
send_email(f"Error: {error_info['error']}", error_info)
# Nutzung
notifier = ErrorNotifier()
notifier.add_notifier(slack_notifier)
notifier.add_notifier(email_notifier)
try:
critical_operation()
except Exception as e:
notifier.notify(e, context={'operation': 'critical_operation'})