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_users

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)

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 client

Se 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 True

Supprime 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 True

Supprime 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 True

Supprime 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 True

Supprime 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, users

Ré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