State Machine para Automações: O Padrão Que Impede Seus Scripts de Virarem Espaguete (Python Puro)

PorOlivetto

Você já teve aquele script de automação que funcionava perfeitamente até o servidor reiniciar no meio da execução? E quando voltou, ninguém sabia em que ponto tinha parado — se tinha enviado o email, se tinha gravado no banco, se tinha travado no meio do webhook?

State machine não é frescura de arquiteto. É a diferença entre uma automação que se recupera sozinha e uma que precisa de você às 3 da manhã resolvendo estado inconsistente com script ad-hoc que você escreveu dormindo.

Hoze eu vou te mostrar como construir uma máquina de estados para automações em Python puro — sem bibliotecas pesadas, sem framework, sem complicação. Só código que funciona, persiste estado em arquivo local e sobrevive a quedas de energia.

Painel de monitoramento de automações com dashboard de estados e transições em tempo real
Painel de estados: cada automação é um nó navegando entre transições definidas

O Problema Que Todo Mundo Ignora (Até Que Explode)

A maioria das automações que eu vejo por aí segue o mesmo padrão:

# O clássico script linear que todo mundo escreve
def executar_automacao():
    passo_1()  # baixa dados da API
    passo_2()  # processa os dados
    passo_3()  # salva no banco
    passo_4()  # envia notificação
    passo_5()  # gera relatório

Parece inofensivo, né? Até que o passo 3 falha.

O que acontece?

  • Os dados foram baixados (passo 1 ✅)
  • Foram processados (passo 2 ✅)
  • Mas não foram salvos (passo 3 ❌)
  • E o passo 4 nem executou

Agora você reexecuta o script. Ele baixa os dados de novo, processa de novo, e talvez — dependendo do timing — duplique registros ou gere relatórios inconsistentes. Ou pior: o passo 4 envia uma notificação que já foi enviada antes porque você não tinha como saber.

Isso não é bug. É falta de controle de estado. E state machine resolve isso de uma vez.

O Que É Uma State Machine (Sem Bla-Bla-Bla Acadêmico)

Uma máquina de estados é, na prática, um conceito simples:

Sua automação sempre está em UM estado definido. Ela só muda de estado quando uma transição explícita acontece. E você sabe exatamente em que ponto parou.

Pensando nos estados da nossa automação de exemplo:

  • IDLE → esperando execução
  • FETCHING → baixando dados
  • PROCESSING → transformando dados
  • PERSISTING → salvando no banco
  • NOTIFYING → enviando alertas
  • REPORTING → gerando relatório
  • COMPLETED → terminou com sucesso
  • FAILED → erro em algum ponto
  • RECOVERING → tentando se recuperar

Cada estado tem transições válidas. Do IDLE você vai para FETCHING. Do FETCHING, só pode ir para PROCESSING (sucesso) ou FAILED (erro). Não tem como pular direto para COMPLETED sem passar por tudo antes.

Implementação Prática: State Machine em Python Puro

Vamos construir algo que funciona de verdade. Primeiro, a base:

from enum import Enum, auto
from typing import Callable, Dict, Set
from datetime import datetime
import json
import os

class Estado(Enum):
    IDLE = auto()
    FETCHING = auto()
    PROCESSING = auto()
    PERSISTING = auto()
    NOTIFYING = auto()
    REPORTING = auto()
    COMPLETED = auto()
    FAILED = auto()
    RECOVERING = auto()

class TransicaoInvalida(Exception):
    """Tentativa de transição que não está definida"""
    pass

class StateMachine:
    def __init__(self, estado_inicial: Estado, nome: str = "automacao"):
        self.nome = nome
        self.estado_atual = estado_inicial
        self.historico: list[dict] = []
        self._transicoes: Dict[Estado, Set[Estado]] = {}
        self._arquivo_estado = f"/tmp/{nome}_estado.json"
        self._definir_transicoes_padrao()

    def _definir_transicoes_padrao(self):
        """Define quais transições são válidas para cada estado"""
        self._transicoes = {
            Estado.IDLE: {Estado.FETCHING, Estado.FAILED},
            Estado.FETCHING: {Estado.PROCESSING, Estado.FAILED},
            Estado.PROCESSING: {Estado.PERSISTING, Estado.FAILED},
            Estado.PERSISTING: {Estado.NOTIFYING, Estado.FAILED},
            Estado.NOTIFYING: {Estado.REPORTING, Estado.FAILED},
            Estado.REPORTING: {Estado.COMPLETED, Estado.FAILED},
            Estado.FAILED: {Estado.RECOVERING, Estado.IDLE},
            Estado.RECOVERING: {Estado.FETCHING, Estado.PROCESSING, Estado.PERSISTING},
            Estado.COMPLETED: set(),  # estado terminal
        }

    def pode_transicionar(self, destino: Estado) -> bool:
        """Verifica se a transição é válida sem executar"""
        permitidas = self._transicoes.get(self.estado_atual, set())
        return destino in permitidas

    def transicionar(self, destino: Estado, contexto: str = ""):
        """Executa a transição de estado com validação"""
        if not self.pode_transicionar(destino):
            raise TransicaoInvalida(
                f"[{self.nome}] Transição inválida: "
                f"{self.estado_atual.name} -> {destino.name}. "
                f"Transições permitidas: {[e.name for e in self._transicoes.get(self.estado_atual, set())]}"
            )

        anterior = self.estado_atual
        self.estado_atual = destino
        self._registrar_historico(anterior, destino, contexto)
        self._persistir_estado()
        print(f"[{self.nome}] {anterior.name} -> {destino.name} {contexto}")

    def _registrar_historico(self, de: Estado, para: Estado, contexto: str):
        self.historico.append({
            "de": de.name,
            "para": para.name,
            "contexto": contexto,
            "timestamp": datetime.now().isoformat(),
        })

    def _persistir_estado(self):
        """Salva o estado atual em arquivo local para recuperação"""
        snapshot = {
            "nome": self.nome,
            "estado_atual": self.estado_atual.name,
            "historico": self.historico,
            "ultima_atualizacao": datetime.now().isoformat(),
        }
        with open(self._arquivo_estado, "w") as f:
            json.dump(snapshot, f, indent=2)

    @classmethod
    def restaurar(cls, nome: str = "automacao"):
        """Restaura estado de arquivo persistente"""
        arquivo = f"/tmp/{nome}_estado.json"
        if not os.path.exists(arquivo):
            sm = cls(Estado.IDLE, nome)
            return sm

        with open(arquivo) as f:
            dados = json.load(f)

        estado = Estado[dados["estado_atual"]]
        sm = cls(estado, nome)
        sm.historico = dados.get("historico", [])
        return sm

Isso aqui é a base. Valida transições, registra histórico e persiste estado em disco. Se o processo morrer no meio, você restaura exatamente de onde parou.

Terminal de comandos mostrando monitoramento de estados de automações em execução
Cada linha é uma transição validada — sem surpresas, sem estado perdido

Orquestrando a Automação Real

Agora vamos usar essa máquina de estados para orquestrar nossa automação de exemplo:

class AutomacaoPipeline:
    def __init__(self, nome: str = "pipeline_dados"):
        self.sm = StateMachine(Estado.IDLE, nome)
        self.dados_baixados = None
        self.dados_processados = None

    def executar(self):
        """Executa o pipeline completo com controle de estado"""
        try:
            # Transição: IDLE -> FETCHING
            self.sm.transicionar(Estado.FETCHING, "iniciando pipeline")
            self._baixar_dados()

            # Transição: FETCHING -> PROCESSING
            self.sm.transicionar(Estado.PROCESSING, "dados recebidos")
            self._processar_dados()

            # Transição: PROCESSING -> PERSISTING
            self.sm.transicionar(Estado.PERSISTING, "dados transformados")
            self._persistir_dados()

            # Transição: PERSISTING -> NOTIFYING
            self.sm.transicionar(Estado.NOTIFYING, "dados salvos")
            self._enviar_notificacao()

            # Transição: NOTIFYING -> REPORTING
            self.sm.transicionar(Estado.REPORTING, "notificação enviada")
            self._gerar_relatorio()

            # Transição: REPORTING -> COMPLETED
            self.sm.transicionar(Estado.COMPLETED, "pipeline finalizado")

        except Exception as e:
            self._tratar_erro(e)

    def _baixar_dados(self):
        # Simula fetch de API externa
        import time
        time.sleep(0.1)  # aqui seria requests.get()
        self.dados_baixados = {"registros": 1500}

    def _processar_dados(self):
        # Simula transformação
        self.dados_processados = {
            "validos": int(self.dados_baixados["registros"] * 0.95),
            "invalidos": int(self.dados_baixados["registros"] * 0.05),
        }

    def _persistir_dados(self):
        # Simula write no banco
        pass

    def _enviar_notificacao(self):
        # Simula envio de alerta
        pass

    def _gerar_relatorio(self):
        # Simula geração de relatório
        pass

    def _tratar_erro(self, erro: Exception):
        """Registra o erro e tenta recuperar"""
        estado_antes_erro = self.sm.estado_atual
        self.sm.transicionar(Estado.FAILED, f"erro: {erro}")
        print(f"  [recuperação] tentando retomar de {estado_antes_erro.name}")

        # Dependendo do estado, recupera de um ponto diferente
        if estado_antes_erro in (Estado.FETCHING, Estado.PROCESSING):
            self.sm.transicionar(Estado.RECOVERING, "fallback para fetching")
            self.sm.transicionar(Estado.FETCHING, "retry após recuperação")
            self._baixar_dados()
            self.sm.transicionar(Estado.PROCESSING, "retry bem-sucedido")
            # ... continua a partir daqui
        else:
            print("  [erro] estado não recuperável automaticamente")

Note o que acontece aqui: cada passo é uma transição explícita. Se qualquer coisa falha, o estado vai para FAILED e o sistema sabe exatamente onde parou. Não tem adivinhação. Não tem “será que o passo 3 rodou?”.

O Box do Perrengue Real

🔧 O Perrengue: Eu implementei isso em produção depois de um deploy que matou o container no meio de um sync de 40 mil registros. O script anterior — sem state machine — ficou 3 dias processando duplicata porque ninguém sabia quantos registros já tinham sido inseridos. Com state machine, o container voltou, leu o arquivo de estado, descobriu que estava em PERSISTING com 23.847 registros salvos, e continuou do registro 23.848. Zero duplicata. Zero intervenção manual.

Por Que Isso É Melhor Que if/else?

Você pode estar pensando: “isso tudo é overkill, eu uso if e try/except“. Eu também pensava assim. Até o dia em que:

  • Um except genérico engoliu um erro de rede e o script continuou processando dados vazios
  • Uma variável de estado ficou como True mas o processo já tinha morrido
  • Dois scripts rodaram ao mesmo tempo e ambos acharam que eram o único ativo

State machine resolve esses problemas porque:

  1. Transições são explícitas — não tem como pular etapa
  2. Estado é persistido — sobrevive a crashes e reinícios
  3. Histórico é auditável — você sabe exatamente o que aconteceu e quando
  4. Validação é automática — tentativa de transição inválida gera exceção imediata
Painel de controle de automação industrial mostrando fluxos e estados de processos automatizados
Assim como painéis industriais monitoram cada etapa, sua automação merece o mesmo controle

Adicionando Timeouts e Guard Clauses

Uma state machine de verdade precisa de timeouts. Se o estado FETCHING durar mais de 30 segundos, algo está errado:

import time

class StateMachineComTimeout(StateMachine):
    def __init__(self, estado_inicial: Estado, nome: str = "automacao",
                 timeout_segundos: int = 30):
        super().__init__(estado_inicial, nome)
        self.timeout = timeout_segundos
        self._inicio_estado = None

    def transicionar(self, destino: Estado, contexto: str = ""):
        # Verifica timeout do estado anterior
        if self._inicio_estado is not None:
            decorrido = time.time() - self._inicio_estado
            if decorrido > self.timeout:
                print(f"  [timeout] estado {self.estado_atual.name} "
                      f"durou {decorrido:.1f}s (limite: {self.timeout}s)")

        super().transicionar(destino, contexto)
        self._inicio_estado = time.time()

Simples. Efetivo. E evita aquela situação onde seu script fica travado em PROCESSING por 4 horas porque a API externa entrou em modo zombie.

Integrando com Systemd e Logs

Para rodar em produção, você não quer só persistir em arquivo — quer logs estruturados:

import logging

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(name)s] %(message)s",
    handlers=[
        logging.FileHandler("/var/log/automente/state_machine.log"),
        logging.StreamHandler(),
    ]
)

logger = logging.getLogger("state_machine")

# No StateMachine, substitua o print por:
# logger.info(f"[{self.nome}] {anterior.name} -> {destino.name} {contexto}")

Com isso, cada transição gera um log com timestamp. Você pode grep, monitorar com fail2ban, enviar para um dashboard — o céu é o limite.

Quando NÃO Usar State Machine

Vamos ser honestos: state machine não é bala de prata. Não use quando:

  • O fluxo tem 2 passos e nunca vai crescer — um try/except resolve
  • Você precisa de paralelismo real — state machine é linear por natureza (embora você possa ter múltiplas instâncias)
  • O estado depende de condições externas complexas — talvez um event-driven architecture faça mais sentido

Use quando o custo de falhar no meio do processo é maior que o custo de escrever o código extra. Na minha experiência, esse ponto é atingido na automação #3 ou #4.

Conclusão: Controle é Liberdade

State machine para automações não é sobre ser fancy. É sobre dormir tranquilo sabendo que suas automações sabem onde estão, para onde vão, e o que fazer se algo der errado.

O código completo está acima — copie, adapte, use. Sem dependências externas, sem framework, sem complicação. Só Python puro resolvendo um problema real.

Qual automação sua precisa desse controle? Me conta nos comentários — ou melhor, me diz aqui embaixo: qual pipeline de dados, qual sync, qual processo batch você quer ver implementado com state machine no próximo artigo?

Leitura relacionada:

Posts Similares