Module demo_api.scripts.quick_cleanup
Script de nettoyage simplifié pour les VMs et utilisateurs avec Typer et Rich Utilise des fonctions modulaires pour une meilleure lisibilité
Functions
def cleanup(base_url: str | None = <typer.models.OptionInfo object>,
email: str | None = <typer.models.OptionInfo object>,
password: str | None = <typer.models.OptionInfo object>,
real: bool = <typer.models.OptionInfo object>,
delay: float = <typer.models.OptionInfo object>) ‑> None-
Expand source code
@app.command() def cleanup( base_url: Optional[str] = typer.Option( None, "--base-url", "-u", help="URL de base de l'API" ), email: Optional[str] = typer.Option( None, "--email", "-e", help="Email pour l'authentification" ), password: Optional[str] = typer.Option( None, "--password", "-p", help="Mot de passe pour l'authentification" ), real: bool = typer.Option( False, "--real", "-r", help="Effectue la suppression réelle" ), delay: float = typer.Option( 0, "--delay", "-d", help="Délai en secondes entre les opérations" ), ) -> None: """ Script de nettoyage pour les VMs et utilisateurs 💡 Exemples d'usage: • Mode simulation (par défaut): python quick_cleanup_simplified.py • Suppression réelle: python quick_cleanup_simplified.py --real • Avec délai personnalisé: python quick_cleanup_simplified.py --real --delay 3 """ simulate = not real quick_cleanup(base_url, email, password, simulate, delay)Script de nettoyage pour les VMs et utilisateurs
💡 Exemples d'usage:
• Mode simulation (par défaut): python quick_cleanup_simplified.py
• Suppression réelle: python quick_cleanup_simplified.py –real
• Avec délai personnalisé: python quick_cleanup_simplified.py –real –delay 3
def cleanup_data(client, vms: list, users: list, delay: float) ‑> Tuple[int, int]-
Expand source code
def cleanup_data(client, vms: list, users: list, delay: float) -> Tuple[int, int]: """Logique métier principale de nettoyage avec barre de progression globale Args: client: Client API vms: Liste des VMs users: Liste des utilisateurs delay: Délai entre suppressions Returns: Tuple (deleted_vms, deleted_users) """ total_items = len(vms) + len(users) if total_items == 0: console.print("[yellow]⚠️ Aucun élément à supprimer[/yellow]") return 0, 0 # Affichage du résumé global console.print( Panel.fit( f"[bold cyan]🚀 DÉBUT DU NETTOYAGE[/bold cyan]\n" f"Total: [bold]{total_items}[/bold] éléments à supprimer\n" f"VMs: [bold]{len(vms)}[/bold] | Utilisateurs: [bold]{len(users)}[/bold]", border_style="cyan", ) ) console.print() # Barre de progression globale with Progress( SpinnerColumn(), TextColumn("[progress.description]{task.description}"), BarColumn(bar_width=None), MofNCompleteColumn(), TextColumn("•"), TimeElapsedColumn(), TextColumn("•"), TimeRemainingColumn(), console=console, expand=True, ) as global_progress: global_task = global_progress.add_task("Nettoyage global", total=total_items) # Suppression des VMs deleted_vms = delete_items_with_progress_and_global( client, vms, "vm", delay, global_progress, global_task ) # Suppression des utilisateurs deleted_users = delete_items_with_progress_and_global( client, users, "user", delay, global_progress, global_task ) return deleted_vms, deleted_usersLogique métier principale de nettoyage avec barre de progression globale
Args
client- Client API
vms- Liste des VMs
users- Liste des utilisateurs
delay- Délai entre suppressions
Returns
Tuple (deleted_vms, deleted_users)
def connect_to_api(base_url: str | None, email: str | None, password: str | None)-
Expand source code
def connect_to_api( base_url: Optional[str], email: Optional[str], password: Optional[str] ): """Se connecte à l'API et retourne le client""" with console.status("[bold green]Connexion à l'API..."): client = create_authenticated_client(base_url, email, password) return clientSe connecte à l'API et retourne le client
def delete_items_batch(client, items: list, item_type: str, delay: float) ‑> int-
Expand source code
def delete_items_batch(client, items: list, item_type: str, delay: float) -> int: """Supprime une liste d'éléments avec gestion des pauses Args: client: Client API items: Liste des éléments à supprimer item_type: Type d'élément ('vm' ou 'user') delay: Délai entre suppressions Returns: Nombre d'éléments supprimés """ if not items: return 0 for i, item in enumerate(items): # Suppression selon le type if item_type == "vm": delete_vm_data(client, item) else: # user delete_user_data(client, item) # Pause si pas le dernier élément if i < len(items) - 1: time.sleep(delay) # Pause supplémentaire avant prochaine section if item_type == "vm": time.sleep(delay + 1) return len(items)Supprime une liste d'éléments avec gestion des pauses
Args
client- Client API
items- Liste des éléments à supprimer
item_type- Type d'élément ('vm' ou 'user')
delay- Délai entre suppressions
Returns
Nombre d'éléments supprimés
def delete_items_with_progress(client, items: list, item_type: str, delay: float) ‑> int-
Expand source code
def delete_items_with_progress( client, items: list, item_type: str, delay: float ) -> int: """Supprime des éléments avec barre de progression détaillée et affichage""" if not items: console.print(f"[yellow]⚠️ Aucun{item_type} à supprimer[/yellow]") return 0 # Affichage du panneau de suppression display_deletion_progress(item_type, delay) # Calcul du temps estimé total (suppression + délais) estimated_time = len(items) * delay + (len(items) - 1) * delay console.print(f"[dim]⏱️ Temps estimé: ~{estimated_time:.1f}s[/dim]") console.print() with Progress( SpinnerColumn(), TextColumn("[progress.description]{task.description}"), BarColumn(bar_width=None), MofNCompleteColumn(), TextColumn("•"), TimeElapsedColumn(), TextColumn("•"), TimeRemainingColumn(), console=console, expand=True, ) as progress: task = progress.add_task(f"Suppression des {item_type}s", total=len(items)) for i, item in enumerate(items): # Mise à jour de la description avec le nom de l'élément item_name = item.get("name", f"ID {item['id']}") progress.update( task, description=f"Suppression {item_type}: {item_name}", advance=0 ) # Suppression avec affichage if item_type == "vm": delete_single_vm_with_display(client, item) else: # user delete_single_user_with_display(client, item) progress.update(task, advance=1) # Pause si pas le dernier élément if i < len(items) - 1: progress.update( task, description=f"Pause {delay}s avant le prochain {item_type}..." ) time.sleep(delay) # Affichage du résultat display_deletion_result(item_type, len(items), len(items)) # Pause supplémentaire avant prochaine section if item_type == "vm": display_pause_message(delay + 1, "Pause avant les utilisateurs") return len(items)Supprime des éléments avec barre de progression détaillée et affichage
def delete_items_with_progress_and_global(client, items: list, item_type: str, delay: float, global_progress, global_task) ‑> int-
Expand source code
def delete_items_with_progress_and_global( client, items: list, item_type: str, delay: float, global_progress, global_task ) -> int: """Supprime des éléments avec barre de progression détaillée et mise à jour globale""" if not items: return 0 # Affichage du panneau de suppression display_deletion_progress(item_type, delay) # Calcul du temps estimé total (suppression + délais) estimated_time = len(items) * delay + (len(items) - 1) * delay console.print(f"[dim]⏱️ Temps estimé: ~{estimated_time:.1f}s[/dim]") console.print() for i, item in enumerate(items): # Mise à jour de la barre globale avec le nom de l'élément item_name = item.get("name", f"ID {item['id']}") global_progress.update( global_task, description=f"Suppression {item_type}: {item_name}", advance=0 ) # Suppression avec affichage if item_type == "vm": delete_single_vm_with_display(client, item) else: # user delete_single_user_with_display(client, item) global_progress.update(global_task, advance=1) # Pause si pas le dernier élément if i < len(items) - 1: global_progress.update( global_task, description=f"Pause {delay}s avant le prochain {item_type}...", ) time.sleep(delay) # Affichage du résultat display_deletion_result(item_type, len(items), len(items)) # Pause supplémentaire avant prochaine section if item_type == "vm": display_pause_message(delay + 1, "Pause avant les utilisateurs") return len(items)Supprime des éléments avec barre de progression détaillée et mise à jour globale
def delete_single_user_with_display(client, user: dict) ‑> bool-
Expand source code
def delete_single_user_with_display(client, user: dict) -> bool: """Supprime un utilisateur avec affichage""" with console.status(f"Suppression utilisateur {user['id']}: {user['name']}..."): client.users.delete_user(user["id"]) display_success_message("Utilisateur", user["name"]) return TrueSupprime un utilisateur avec affichage
def delete_single_vm_with_display(client, vm: dict) ‑> bool-
Expand source code
def delete_single_vm_with_display(client, vm: dict) -> bool: """Supprime une VM avec affichage""" with console.status(f"Suppression VM {vm['id']}: {vm['name']}..."): client.vms.delete(vm["id"]) display_success_message("VM", vm["name"]) return TrueSupprime une VM avec affichage
def delete_user_data(client, user: dict) ‑> bool-
Expand source code
def delete_user_data(client, user: dict) -> bool: """Supprime un utilisateur (logique pure sans affichage)""" client.users.delete_user(user["id"]) return TrueSupprime un utilisateur (logique pure sans affichage)
def delete_vm_data(client, vm: dict) ‑> bool-
Expand source code
def delete_vm_data(client, vm: dict) -> bool: """Supprime une VM (logique pure sans affichage)""" client.vms.delete(vm["id"]) return TrueSupprime une VM (logique pure sans affichage)
def display_api_config(client) ‑> None-
Expand source code
def display_api_config(client) -> None: """Affiche la configuration de l'API""" config_table = Table(title="🔗 Configuration API") config_table.add_column("Paramètre", style="cyan") config_table.add_column("Valeur", style="magenta") config_table.add_row("Base URL", client.base_url) config_table.add_row( "Authentifié", "✅ Oui" if client.is_authenticated() else "❌ Non" ) console.print(config_table) console.print()Affiche la configuration de l'API
def display_deletion_progress(item_type: str, delay: float) ‑> None-
Expand source code
def display_deletion_progress(item_type: str, delay: float) -> None: """Affiche le panneau de suppression""" console.print( Panel.fit( f"[bold red]🗑️ Suppression des {item_type}s...[/bold red]\n" f"Délai: [bold]{delay}s[/bold]", border_style="red", ) )Affiche le panneau de suppression
def display_deletion_result(item_type: str, deleted_count: int, total_count: int) ‑> None-
Expand source code
def display_deletion_result( item_type: str, deleted_count: int, total_count: int ) -> None: """Affiche le résultat de suppression""" console.print( f"[bold cyan]📊 {item_type}s supprimés: [green]{deleted_count}/{total_count}[/green][/bold cyan]" )Affiche le résultat de suppression
def display_header(simulate: bool) ‑> None-
Expand source code
def display_header(simulate: bool) -> None: """Affiche l'en-tête selon le mode""" if simulate: console.print( Panel.fit( "[bold blue]🧹 MODE SIMULATION[/bold blue]\n" "Aucune donnée ne sera supprimée", border_style="blue", ) ) else: console.print( Panel.fit( "[bold red]🗑️ MODE SUPPRESSION RÉELLE[/bold red]\n" "⚠️ TOUTES LES DONNÉES SERONT SUPPRIMÉES !", border_style="red", ) )Affiche l'en-tête selon le mode
def display_operation_config(delay: float, simulate: bool) ‑> None-
Expand source code
def display_operation_config(delay: float, simulate: bool) -> None: """Affiche la configuration des opérations""" config_table = Table(title="🔧 Configuration") config_table.add_column("Paramètre", style="cyan") config_table.add_column("Valeur", style="magenta") config_table.add_row("Délai entre opérations", f"{delay}s") config_table.add_row("Mode", "Simulation" if simulate else "Suppression réelle") console.print(config_table) console.print()Affiche la configuration des opérations
def display_pause_message(delay: float, context: str = '') ‑> None-
Expand source code
def display_pause_message(delay: float, context: str = "") -> None: """Affiche un message de pause""" if context: console.print(f"[dim]⏱️ {context} ({delay}s)...[/dim]") else: console.print(f"[dim]⏱️ Pause de {delay}s...[/dim]")Affiche un message de pause
def display_simulation_message() ‑> None-
Expand source code
def display_simulation_message() -> None: """Affiche le message de simulation""" console.print( Panel.fit( "[bold blue]📋 Mode simulation - aucune suppression réelle[/bold blue]\n" "Utilisez [bold]--real[/bold] pour effectuer les suppressions", border_style="blue", ) )Affiche le message de simulation
def display_success_message(item_type: str, item_name: str) ‑> None-
Expand source code
def display_success_message(item_type: str, item_name: str) -> None: """Affiche un message de succès""" console.print(f"[green]✅ {item_type} supprimé: [bold]{item_name}[/bold][/green]")Affiche un message de succès
def display_users_table(users: list) ‑> None-
Expand source code
def display_users_table(users: list) -> None: """Affiche les utilisateurs dans un tableau""" table = Table(title="👥 Utilisateurs") table.add_column("ID", style="cyan") table.add_column("Nom", style="green") table.add_column("Email", style="yellow") console.print(f"[green]✅ {len(users)} utilisateurs trouvés[/green]") for user in users: table.add_row(str(user["id"]), user["name"], user["email"]) console.print(table)Affiche les utilisateurs dans un tableau
def display_vms_table(vms: list) ‑> None-
Expand source code
def display_vms_table(vms: list) -> None: """Affiche les VMs dans un tableau""" table = Table(title="💻 Machines virtuelles") table.add_column("ID", style="cyan") table.add_column("Nom", style="green") table.add_column("Utilisateur", style="yellow") table.add_column("Status", style="magenta") console.print(f"[green]✅ {len(vms)} VMs trouvées[/green]") for vm in vms: table.add_row( str(vm["id"]), vm["name"], str(vm["user_id"]), vm.get("status", "Inconnu") ) console.print(table)Affiche les VMs dans un tableau
def fetch_data(client) ‑> Tuple[list, list]-
Expand source code
def fetch_data(client) -> Tuple[list, list]: """Récupère les données VMs et utilisateurs""" # Récupération VMs console.print("[bold cyan]📊 Données actuelles:[/bold cyan]") with console.status("[bold green]Récupérations des VMs..."): vms = client.vms.get() display_vms_table(vms) console.print() # Récupération utilisateurs with console.status("[bold green]Récupération des utilisateurs..."): users = client.users.get() display_users_table(users) console.print() return vms, usersRécupère les données VMs et utilisateurs
def main()-
Expand source code
def main(): """Point d'entrée principal""" app()Point d'entrée principal
def quick_cleanup(base_url: str | None = None,
email: str | None = None,
password: str | None = None,
simulate: bool = True,
delay: float = 2.5) ‑> None-
Expand source code
def quick_cleanup( base_url: Optional[str] = None, email: Optional[str] = None, password: Optional[str] = None, simulate: bool = True, delay: float = 2.5, ) -> None: """Fonction principale orchestrant le nettoyage""" try: # 1. AFFICHAGE - En-tête selon le mode display_header(simulate) # 2. DONNÉES - Connexion et récupération client = connect_to_api(base_url, email, password) display_api_config(client) vms, users = fetch_data(client) # 3. AFFICHAGE - Configuration des opérations display_operation_config(delay, simulate) # 4. LOGIQUE MÉTIER - Si simulation, arrêt ici if simulate: display_simulation_message() return # 5. LOGIQUE MÉTIER - Suppressions réelles deleted_vms, deleted_users = cleanup_data(client, vms, users, delay) # 6. AFFICHAGE - Résumé final show_summary(vms, users, deleted_vms, deleted_users) except Exception as e: console.print(f"[bold red]❌ Erreur critique: {e}[/bold red]") raise typer.Exit(1)Fonction principale orchestrant le nettoyage
def show_summary(vms: list, users: list, deleted_vms: int, deleted_users: int) ‑> None-
Expand source code
def show_summary(vms: list, users: list, deleted_vms: int, deleted_users: int) -> None: """Affiche le résumé final""" total_deleted = deleted_vms + deleted_users summary_table = Table(title="🎯 Résumé du nettoyage") summary_table.add_column("Type", style="cyan") summary_table.add_column("Supprimé", style="green") summary_table.add_column("Total", style="yellow") summary_table.add_row("VMs", str(deleted_vms), str(len(vms))) summary_table.add_row("Utilisateurs", str(deleted_users), str(len(users))) summary_table.add_row( "**TOTAL**", f"[bold]{total_deleted}[/bold]", f"[bold]{len(vms) + len(users)}[/bold]", ) console.print(summary_table) console.print( Panel.fit( "[bold green]✅ NETTOYAGE TERMINÉ AVEC SUCCÈS ![/bold green]", border_style="green", ) )Affiche le résumé final