162 lines
7.4 KiB
Python
162 lines
7.4 KiB
Python
"""
|
|
Schémas Pydantic pour les schedules - modèles API complets.
|
|
"""
|
|
|
|
from datetime import datetime, timezone
|
|
from typing import Optional, List, Literal, Dict, Any
|
|
import uuid
|
|
|
|
from pydantic import BaseModel, Field, ConfigDict, field_validator
|
|
import pytz
|
|
|
|
|
|
class ScheduleRecurrence(BaseModel):
|
|
"""Configuration de récurrence pour un schedule."""
|
|
type: Literal["daily", "weekly", "monthly", "custom"] = "daily"
|
|
time: str = Field(default="02:00", description="Heure d'exécution HH:MM")
|
|
days: Optional[List[int]] = Field(default=None, description="Jours de la semaine (1-7, lundi=1) pour weekly")
|
|
day_of_month: Optional[int] = Field(default=None, ge=1, le=31, description="Jour du mois (1-31) pour monthly")
|
|
cron_expression: Optional[str] = Field(default=None, description="Expression cron pour custom")
|
|
|
|
|
|
class Schedule(BaseModel):
|
|
"""Modèle complet d'un schedule pour l'API."""
|
|
id: str = Field(default_factory=lambda: f"sched_{uuid.uuid4().hex[:12]}")
|
|
name: str = Field(..., min_length=3, max_length=100, description="Nom du schedule")
|
|
description: Optional[str] = Field(default=None, max_length=500)
|
|
playbook: str = Field(..., description="Nom du playbook à exécuter")
|
|
target_type: Literal["group", "host"] = Field(default="group", description="Type de cible")
|
|
target: str = Field(default="all", description="Nom du groupe ou hôte cible")
|
|
extra_vars: Optional[Dict[str, Any]] = Field(default=None, description="Variables supplémentaires")
|
|
schedule_type: Literal["once", "recurring"] = Field(default="recurring")
|
|
recurrence: Optional[ScheduleRecurrence] = Field(default=None)
|
|
timezone: str = Field(default="America/Montreal", description="Fuseau horaire")
|
|
start_at: Optional[datetime] = Field(default=None, description="Date de début (optionnel)")
|
|
end_at: Optional[datetime] = Field(default=None, description="Date de fin (optionnel)")
|
|
next_run_at: Optional[datetime] = Field(default=None, description="Prochaine exécution calculée")
|
|
last_run_at: Optional[datetime] = Field(default=None, description="Dernière exécution")
|
|
last_status: Literal["success", "failed", "running", "never"] = Field(default="never")
|
|
enabled: bool = Field(default=True, description="Schedule actif ou en pause")
|
|
retry_on_failure: int = Field(default=0, ge=0, le=3, description="Nombre de tentatives en cas d'échec")
|
|
timeout: int = Field(default=3600, ge=60, le=86400, description="Timeout en secondes")
|
|
notification_type: Literal["none", "all", "errors"] = Field(default="all", description="Type de notification")
|
|
tags: List[str] = Field(default_factory=list, description="Tags pour catégorisation")
|
|
run_count: int = Field(default=0, description="Nombre total d'exécutions")
|
|
success_count: int = Field(default=0, description="Nombre de succès")
|
|
failure_count: int = Field(default=0, description="Nombre d'échecs")
|
|
created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
|
|
updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
|
|
|
|
model_config = ConfigDict(
|
|
json_encoders={datetime: lambda v: v.isoformat() if v else None}
|
|
)
|
|
|
|
@field_validator('recurrence', mode='before')
|
|
@classmethod
|
|
def validate_recurrence(cls, v, info):
|
|
return v
|
|
|
|
|
|
class ScheduleRun(BaseModel):
|
|
"""Historique d'une exécution de schedule."""
|
|
id: str = Field(default_factory=lambda: f"run_{uuid.uuid4().hex[:12]}")
|
|
schedule_id: str = Field(..., description="ID du schedule parent")
|
|
task_id: Optional[str] = Field(default=None, description="ID de la tâche créée")
|
|
started_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
|
|
finished_at: Optional[datetime] = Field(default=None)
|
|
status: Literal["running", "success", "failed", "canceled"] = Field(default="running")
|
|
duration_seconds: Optional[float] = Field(default=None)
|
|
hosts_impacted: int = Field(default=0)
|
|
error_message: Optional[str] = Field(default=None)
|
|
retry_attempt: int = Field(default=0, description="Numéro de la tentative (0 = première)")
|
|
|
|
model_config = ConfigDict(
|
|
json_encoders={datetime: lambda v: v.isoformat() if v else None}
|
|
)
|
|
|
|
|
|
class ScheduleCreateRequest(BaseModel):
|
|
"""Requête de création d'un schedule."""
|
|
name: str = Field(..., min_length=3, max_length=100)
|
|
description: Optional[str] = Field(default=None, max_length=500)
|
|
playbook: str = Field(...)
|
|
target_type: Literal["group", "host"] = Field(default="group")
|
|
target: str = Field(default="all")
|
|
extra_vars: Optional[Dict[str, Any]] = Field(default=None)
|
|
schedule_type: Literal["once", "recurring"] = Field(default="recurring")
|
|
recurrence: Optional[ScheduleRecurrence] = Field(default=None)
|
|
timezone: str = Field(default="America/Montreal")
|
|
start_at: Optional[datetime] = Field(default=None)
|
|
end_at: Optional[datetime] = Field(default=None)
|
|
enabled: bool = Field(default=True)
|
|
retry_on_failure: int = Field(default=0, ge=0, le=3)
|
|
timeout: int = Field(default=3600, ge=60, le=86400)
|
|
notification_type: Literal["none", "all", "errors"] = Field(default="all")
|
|
tags: List[str] = Field(default_factory=list)
|
|
|
|
@field_validator('timezone')
|
|
@classmethod
|
|
def validate_timezone(cls, v: str) -> str:
|
|
try:
|
|
pytz.timezone(v)
|
|
return v
|
|
except pytz.exceptions.UnknownTimeZoneError:
|
|
raise ValueError(f"Fuseau horaire invalide: {v}")
|
|
|
|
|
|
class ScheduleUpdateRequest(BaseModel):
|
|
"""Requête de mise à jour d'un schedule."""
|
|
name: Optional[str] = Field(default=None, min_length=3, max_length=100)
|
|
description: Optional[str] = Field(default=None, max_length=500)
|
|
playbook: Optional[str] = Field(default=None)
|
|
target_type: Optional[Literal["group", "host"]] = Field(default=None)
|
|
target: Optional[str] = Field(default=None)
|
|
extra_vars: Optional[Dict[str, Any]] = Field(default=None)
|
|
schedule_type: Optional[Literal["once", "recurring"]] = Field(default=None)
|
|
recurrence: Optional[ScheduleRecurrence] = Field(default=None)
|
|
timezone: Optional[str] = Field(default=None)
|
|
start_at: Optional[datetime] = Field(default=None)
|
|
end_at: Optional[datetime] = Field(default=None)
|
|
enabled: Optional[bool] = Field(default=None)
|
|
retry_on_failure: Optional[int] = Field(default=None, ge=0, le=3)
|
|
timeout: Optional[int] = Field(default=None, ge=60, le=86400)
|
|
notification_type: Optional[Literal["none", "all", "errors"]] = Field(default=None)
|
|
tags: Optional[List[str]] = Field(default=None)
|
|
|
|
|
|
class ScheduleStats(BaseModel):
|
|
"""Statistiques globales des schedules."""
|
|
total: int = 0
|
|
active: int = 0
|
|
paused: int = 0
|
|
expired: int = 0
|
|
next_execution: Optional[datetime] = None
|
|
next_schedule_name: Optional[str] = None
|
|
failures_24h: int = 0
|
|
executions_24h: int = 0
|
|
success_rate_7d: float = 0.0
|
|
|
|
|
|
class ScheduleListResponse(BaseModel):
|
|
"""Réponse API pour la liste des schedules."""
|
|
schedules: List[dict] = Field(default_factory=list)
|
|
count: int = 0
|
|
|
|
|
|
class UpcomingExecution(BaseModel):
|
|
"""Prochaine exécution planifiée."""
|
|
schedule_id: str
|
|
schedule_name: str
|
|
playbook: str
|
|
target: str
|
|
next_run_at: Optional[str] = None
|
|
tags: List[str] = Field(default_factory=list)
|
|
|
|
|
|
class CronValidationResult(BaseModel):
|
|
"""Résultat de validation d'une expression cron."""
|
|
valid: bool
|
|
expression: str
|
|
next_runs: Optional[List[str]] = None
|
|
error: Optional[str] = None
|