Dizionari Python per gestire workflow e transizioni di stato

by theArchitect
SHARE
Dizionari Python per gestire workflow e transizioni di stato
© Guida-HTML5.it

Per gestire un workflow in Python, associa ogni stato agli stati successivi consentiti usando un dizionario di insiemi. La struttura rende esplicite le regole e permette di rifiutare transizioni non valide prima di modificare i dati. In questo esempio, un ticket conserva lo stato corrente e una cronologia con timestamp UTC: un approccio adatto a servizi, API e processi di approvazione di dimensioni contenute.

Sintassi rapida (TL;DR):

transizioni = {"nuova": {"assegnata", "annullata"}, "assegnata": {"risolta"}}
stato = "nuova"
nuovo_stato = "assegnata"

if nuovo_stato not in transizioni.get(stato, set()):
    raise ValueError(f"Transizione non valida: {stato} -> {nuovo_stato}")

stato = nuovo_stato

Quando usare un dizionario Python per validare le transizioni di stato

Un ticket di assistenza, una pratica o un ordine seguono spesso un ciclo di vita con vincoli: una richiesta appena creata può essere assegnata o annullata, ma non chiusa direttamente. Una lunga sequenza di if e elif tende a spargere queste regole nel programma e rende più difficile controllare tutti i percorsi possibili.

Con un dizionario, la chiave rappresenta lo stato corrente e il valore è l’insieme degli stati raggiungibili. Per esempio, "nuova": {"assegnata", "annullata"} dichiara una regola in un punto solo. Recuperare una chiave e verificare l’appartenenza a un insieme hanno costo medio O(1), quindi i controlli sono efficienti anche quando il numero di stati cresce.

Il pattern è indicato quando il workflow è relativamente piccolo, statico e gestito da una singola applicazione. Non sostituisce un motore di workflow o un database: se le regole dipendono da permessi, configurazioni modificabili o transazioni distribuite, occorre aggiungere quei controlli nel livello appropriato.

Implementazione di un workflow per ticket di assistenza in Python

Il programma seguente crea un ticket, applica transizioni consentite e registra ogni cambiamento. Salvalo, per esempio, come workflow.py ed eseguilo con Python 3.11 o superiore. Un orologio iniettabile rende l’esempio riproducibile e facilita i test senza dipendere dall’ora reale.

from collections.abc import Callable
from datetime import datetime, timezone
from typing import TypedDict


class Evento(TypedDict):
    stato: str
    data: str
    nota: str


class Richiesta(TypedDict):
    numero: int
    cliente: str
    descrizione: str
    stato: str
    cronologia: list[Evento]


TRANSIZIONI_CONSENTITE: dict[str, frozenset[str]] = {
    "nuova": frozenset({"assegnata", "annullata"}),
    "assegnata": frozenset({"in_lavorazione", "annullata"}),
    "in_lavorazione": frozenset({"risolta", "annullata"}),
    "risolta": frozenset({"chiusa"}),
    "chiusa": frozenset(),
    "annullata": frozenset(),
}


def timestamp_utc(
    orologio: Callable[[], datetime] | None = None,
) -> str:
    """Restituisce un timestamp ISO 8601 con offset UTC."""
    adesso = (orologio or (lambda: datetime.now(timezone.utc)))()

    if adesso.tzinfo is None or adesso.utcoffset() is None:
        raise ValueError("L'orologio deve restituire una data con fuso orario.")

    return adesso.astimezone(timezone.utc).isoformat(timespec="seconds")


def crea_richiesta(
    numero: int,
    cliente: str,
    descrizione: str,
    *,
    orologio: Callable[[], datetime] | None = None,
) -> Richiesta:
    """Crea un ticket nello stato iniziale e registra l'evento."""
    if numero <= 0:
        raise ValueError("Il numero della richiesta deve essere positivo.")

    cliente = cliente.strip()
    descrizione = descrizione.strip()
    if not cliente or not descrizione:
        raise ValueError("Cliente e descrizione non possono essere vuoti.")

    data = timestamp_utc(orologio)
    return {
        "numero": numero,
        "cliente": cliente,
        "descrizione": descrizione,
        "stato": "nuova",
        "cronologia": [
            {
                "stato": "nuova",
                "data": data,
                "nota": "Richiesta creata",
            }
        ],
    }


def cambia_stato(
    richiesta: Richiesta,
    nuovo_stato: str,
    nota: str = "",
    *,
    orologio: Callable[[], datetime] | None = None,
) -> None:
    """Valida e registra una transizione senza lasciare aggiornamenti parziali."""
    stato_attuale = richiesta["stato"]
    consentiti = TRANSIZIONI_CONSENTITE.get(stato_attuale)

    if consentiti is None:
        raise ValueError(f"Stato corrente sconosciuto: {stato_attuale!r}")
    if nuovo_stato not in TRANSIZIONI_CONSENTITE:
        raise ValueError(f"Stato di destinazione sconosciuto: {nuovo_stato!r}")
    if nuovo_stato not in consentiti:
        elenco = ", ".join(sorted(consentiti)) or "nessuno"
        raise ValueError(
            f"Transizione non valida: {stato_attuale} -> {nuovo_stato}. "
            f"Stati consentiti: {elenco}"
        )

    # Calcola e valida i dati prima di modificare la richiesta.
    data = timestamp_utc(orologio)
    evento: Evento = {
        "stato": nuovo_stato,
        "data": data,
        "nota": nota.strip(),
    }

    richiesta["cronologia"].append(evento)
    richiesta["stato"] = nuovo_stato


def stampa_riepilogo(richiesta: Richiesta) -> None:
    """Stampa lo stato del ticket e la cronologia."""
    print(f"Richiesta #{richiesta['numero']}")
    print(f"Cliente: {richiesta['cliente']}")
    print(f"Descrizione: {richiesta['descrizione']}")
    print(f"Stato attuale: {richiesta['stato']}")
    print("Cronologia:")

    for evento in richiesta["cronologia"]:
        nota = evento["nota"] or "Nessuna nota"
        print(f"- {evento['data']} | {evento['stato']} | {nota}")


def main() -> None:
    # L'orario fisso produce lo stesso output a ogni esecuzione.
    orologio_demo = lambda: datetime(
        2025, 3, 8, 10, 30, tzinfo=timezone.utc
    )
    richiesta = crea_richiesta(
        numero=1042,
        cliente="Luca Bianchi",
        descrizione="Impossibile accedere all'area riservata",
        orologio=orologio_demo,
    )

    cambiamenti = [
        ("assegnata", "Assegnata al team supporto"),
        ("in_lavorazione", "Verifica delle credenziali"),
        ("risolta", "Accesso ripristinato"),
        ("chiusa", "Conferma ricevuta dal cliente"),
    ]
    for stato, nota in cambiamenti:
        cambia_stato(richiesta, stato, nota, orologio=orologio_demo)

    stampa_riepilogo(richiesta)


if __name__ == "__main__":
    main()

Output in Console dell'esecuzione

Richiesta #1042
Cliente: Luca Bianchi
Descrizione: Impossibile accedere all'area riservata
Stato attuale: chiusa
Cronologia:
- 2025-03-08T10:30:00+00:00 | nuova | Richiesta creata
- 2025-03-08T10:30:00+00:00 | assegnata | Assegnata al team supporto
- 2025-03-08T10:30:00+00:00 | in_lavorazione | Verifica delle credenziali
- 2025-03-08T10:30:00+00:00 | risolta | Accesso ripristinato
- 2025-03-08T10:30:00+00:00 | chiusa | Conferma ricevuta dal cliente

Come funzionano dizionario delle transizioni e cronologia del workflow

1. Dizionario di stati e insiemi immutabili

TRANSIZIONI_CONSENTITE definisce il grafo diretto del workflow: una chiave è uno stato di partenza e il relativo frozenset elenca gli stati raggiungibili. Usare un insieme rende esplicito che l’ordine delle destinazioni non conta; usare frozenset impedisce modifiche accidentali alle regole durante l’esecuzione. Uno stato terminale, come "chiusa", ha un insieme vuoto.

2. TypedDict, stato corrente e cronologia degli eventi

TypedDict descrive la forma dei dizionari per strumenti di analisi statica, come mypy e pyright, senza introdurre una nuova classe di oggetti a runtime. La richiesta conserva lo stato corrente per le letture rapide e una lista di eventi per ricostruire le modifiche. Aggiungere un evento costa O(1) ammortizzato; l’intera cronologia occupa memoria proporzionale al numero di transizioni registrate.

3. Validazione preventiva e timestamp UTC

cambia_stato() controlla separatamente lo stato corrente, quello di destinazione e la transizione. Il timestamp viene calcolato prima di aggiornare il dizionario: se l’orologio è invalido, la funzione solleva ValueError senza lasciare stato e cronologia incoerenti. timestamp_utc() rifiuta date senza fuso orario e converte quelle valide in UTC, serializzandole in formato ISO 8601.

Confronto tra dizionario di transizioni, if e librerie di workflow

La scelta dipende dalla dimensione del processo e da dove risiedono le regole. Per un workflow semplice, il dizionario è leggibile e non richiede dipendenze esterne; per requisiti più articolati, una struttura più avanzata può risultare preferibile.

Metodo / Approccio Sintassi / Caratteristica Performance / Complessità Quando sceglierlo
Dizionario e set stato in transizioni.get(corrente, set()) Ricerca media O(1); regole centralizzate Workflow piccoli o medi con regole statiche
Catena di if/elif if corrente == "nuova": ... O(n) nel numero di condizioni; manutenzione più dispersiva Uno o due casi semplici, senza configurazione riutilizzabile
Macchina a stati dedicata Regole, callback e guardie strutturate Dipende dalla libreria e dal modello adottato Workflow complessi, estendibili o con azioni per transizione

Errori comuni da evitare e best practice per workflow in produzione

  • Chiave assente e KeyError: accedere direttamente a transizioni[stato] solleva un errore se lo stato non esiste. Validare lo stato o usare .get() e distinguere una chiave sconosciuta da uno stato terminale.
  • Modificare i dati prima della validazione: aggiornare lo stato e poi controllare la transizione può lasciare la richiesta in una condizione incoerente. Eseguire tutti i controlli e preparare l’evento prima di mutare i dati.
  • Timestamp senza fuso orario: un valore naive non identifica un istante in modo affidabile. Richiedere date timezone-aware e normalizzarle a UTC, soprattutto quando gli eventi vengono confrontati o persistiti.
  • Regole non sincronizzate con i dati: il dizionario protegge le transizioni eseguite attraverso questa funzione, ma non impedisce modifiche dirette a richiesta["stato"]. Incapsulare l’aggiornamento e usare test automatici per ogni transizione consentita e negata.

Domande frequenti sui workflow Python basati su dizionari

Come verificare se una transizione di stato è consentita in Python?

Memorizza le destinazioni ammesse in un dizionario di insiemi e verifica che il nuovo stato appartenga all’insieme associato allo stato corrente. Se la chiave non esiste, lo stato corrente non è valido; se l’insieme è vuoto, lo stato è terminale.

È meglio usare un set o una lista per le transizioni?

Un set è adatto quando serve verificare l’appartenenza e l’ordine delle destinazioni non è rilevante: il costo medio del controllo è O(1). Una lista conserva l’ordine, ma la ricerca di un elemento richiede O(n); sceglila solo se quell’ordine serve davvero.

Come testare un workflow senza dipendere dall’ora di sistema?

Inietta una funzione che restituisce sempre lo stesso datetime timezone-aware, come fa orologio_demo. Nei test puoi verificare lo stato e la cronologia dopo una transizione valida e accertare che una transizione non valida sollevi ValueError senza modificare la richiesta.

Documentazione ufficiale Python su dizionari, tipi e date

SHARE