Django 6.0 Tasks com TDD: três paredes que o caminho feliz esconde

Django 6.0 Tasks com TDD: três paredes que o caminho feliz esconde

O Django 6.0 trouxe um framework de tasks nativo. Segui o tutorial, funcionou de primeira - e foi exatamente aí que desconfiei. Decidi refazer tudo com TDD, escrevendo o teste antes de cada linha. Bati em três paredes que o caminho feliz não mostra.

Este artigo documenta as decisões e os tropeços do cotton-desk-tasks, uma mesa de comercialização de algodão fictícia construída sobre o tutorial de Django Tasks da Real Python. O domínio não é decoração: uma mesa de trading é literalmente um lugar onde trabalho rápido e trabalho lento disputam o mesmo processo - que é o problema que o framework existe pra resolver. (Todos os dados são sintéticos. Nenhum contrato, laudo ou preço real.)

O problema

Entra laudo HVI do laboratório, fecha contrato, ingere índice de preço, gera relatório de posição. Quatro trabalhos com criticidade e latência completamente diferentes:

• Resumir um laudo - o classificador está esperando na tela. Precisa ser rápido.
• Confirmar um contrato - não pode disparar antes do contrato existir no banco.
• Relatório de safra - varre milhares de laudos. Pode esperar.
• Ingestão de preço - uma vez por dia, depois do fechamento do pregão.

Se tudo isso cai numa fila só, um relatório demorado segura a confirmação de um contrato. O framework resolve isso com queue_name e priority:

@task(queue_name="laudos", priority=50)
def resumir_laudo(laudo_id: int) -> str:
    ...

@task(queue_name="relatorios", priority=-10)
def gerar_relatorio_safra(safra: str) -> str:
    ...

Com um worker dedicado por fila (db_worker --queue-name laudos), o isolamento é estrutural: o worker de relatórios fica cego pro que não é dele.

A fronteira: armazenamento permissivo, domínio estrito

Antes das tasks, uma decisão que definiu o resto do projeto: onde mora a regra de negócio.

Um laudo HVI tem quatro parâmetros com faixas comerciais - micronaire entre 3.5 e 4.9, comprimento mínimo de 1.11", resistência mínima de 28 gf/tex, uniformidade mínima de 80%. A tentação é validar isso no save() do model. Mas um laboratório mede o que mede: se o micronaire veio 2.1, isso é um fato que precisa ficar registrado, não um erro a rejeitar.

Então a validação não está no model. Está num value object imutável, sem Django nenhum:

@dataclass(frozen=True)
class HVIParametros:
    micronaire: float
    comprimento: float
    resistencia: float
    uniformidade: float

    def __post_init__(self) -> None:
        self._validar_micronaire()
        self._validar_comprimento()
        self._validar_resistencia()
        self._validar_uniformidade()

E o model expõe uma travessia explícita dessa fronteira:

def to_dominio(self) -> HVIParametros:
    return HVIParametros(
        micronaire=float(self.micronaire),
        ...
    )

O LaudoHVI.objects.create() com micronaire 2.1 salva normalmente. O erro só nasce quando alguém pede o objeto de domínio. O teste que prova isso é o mais importante do projeto:

laudo = LaudoHVI.objects.create(fardo=fardo, micronaire="2.00", ...)   # passa
with pytest.raises(MicronaireForaDaFaixa):
    laudo.to_dominio()   # aqui, sim

Isso paga dividendos depois: quando a task falha, ela falha com desk.domain.MicronaireForaDaFaixa no traceback - não com um ValueError anônimo.

Parede 1: o backend de teste não sabe buscar resultado

O fluxo natural de uma API assíncrona são duas requisições: uma POST enfileira e devolve o id, um GET consulta o status depois. Escrevi o teste, escrevi a view, rodei:

NotImplementedError: This backend does not support retrieving or refreshing results.

O ImmediateBackend - o backend embutido que roda a task inline, ideal pra testes - não implementa get_result(). E o motivo é estrutural, não uma lacuna a ser preenchida: ele nunca persiste nada em lugar nenhum. O único TaskResult que existe é o objeto que o .enqueue() devolveu naquele instante. Buscar por id depois é procurar algo que nunca foi guardado.

Isso quebrou uma premissa que eu tinha embutido no conftest.py. Eu havia forçado ImmediateBackend em toda a suíte, via fixture autouse, pra que os testes não dependessem de um worker rodando em paralelo. Boa decisão - mas errada justamente pra view que existe para o cenário que esse backend não suporta.

A saída foi separar as perguntas. "O backend consegue recuperar o resultado?" é responsabilidade da lib, já testada por ela. "Dado um TaskResult em tal estado, a view devolve o JSON e o HTTP certos?" é a minha - e testa sem banco nem backend:

def test_get_status_da_task_com_laudo_invalido_retorna_falhou(client):
    erro_fake = MagicMock(exception_class_path="desk.domain.MicronaireForaDaFaixa")
    resultado_fake = MagicMock(status=TaskResultStatus.FAILED, errors=[erro_fake])
    with patch("desk.views.default_task_backend.get_result", return_value=resultado_fake):
        resposta = client.get(reverse("status_da_task", args=["qualquer-id"]))
    assert resposta.status_code == 422

422 Unprocessable Entity, não 500: a requisição é processável, o dado é que não é comercialmente válido.

Parede 2: Decimal morre na serialização, e o erro vem antes do worker

Preço é Decimal. Numa mesa de trading isso não é preferência, é obrigação - float em preço é como se paga caro por arredondamento. Então a task de ingestão parecia óbvia:

registrar_leitura_indice.enqueue("ICE-CT2", Decimal("82.35"), "2026-04-10")

TypeError: Unsupported type: <class 'decimal.Decimal'>

O detalhe que importa: o erro estoura dentro do .enqueue(), na serialização dos argumentos, antes de tocar o backend. Não é um problema do banco - é que os argumentos de uma task precisam ser JSON-serializáveis, e a validação acontece na hora de enfileirar. Um Decimal escondido três níveis abaixo num dict só aparece ali.

A correção não é técnica, é de contrato. A assinatura passa a documentar a regra:

@task(queue_name="precos", priority=0)
def registrar_leitura_indice(codigo: str, valor: str, data_pregao: str) -> str:
    """
    `valor` chega como string, não Decimal: argumentos de task passam por serialização JSON
    no `.enqueue()`, e Decimal não sobrevive a esse round-trip - quem chama essa task precisa
    converter antes.
    """
    leitura, _ = IndicePreco.objects.update_or_create(
        codigo=codigo, data_pregao=data_pregao,
        defaults={"valor": Decimal(valor)},
    )

valor: str é o type checker avisando no call site, antes do runtime. A conversão pra Decimal acontece na fronteira de dentro, onde há contexto pra isso.

Parede 3: on_commit nunca dispara no teste - e o mock explode

O gotcha clássico: se a view cria o contrato numa transação e enfileira a confirmação na mesma respiração, o worker pode buscar o contrato antes do commit e não achar nada. A solução é conhecida:

with transaction.atomic():
    contrato = Contrato.objects.create(...)
    transaction.on_commit(partial(confirmar_contrato.enqueue, contrato.id))

O problema é testar isso. Escrevi o teste ingênuo de propósito, mockando o enqueue pra verificar que ele foi chamado:

with patch("desk.views.confirmar_contrato.enqueue") as enqueue_mock:

TypeError: super(type, obj): obj must be an instance or subtype of type

Erro bizarro, causa simples: o decorator @task transforma a função numa instância de Task, que é um dataclass congelado. O unittest.mock.patch funciona fazendo setattr no alvo ao entrar no with e delattr ao sair - e um objeto frozen rejeita as duas operações. Não dá pra mockar método numa Task. (Coerente, aliás: o framework aplicou à Task o mesmo princípio que eu apliquei ao HVIParametros. Objeto que representa um fato não deveria mudar depois de criado.)

Mas o problema maior estava embaixo desse. @pytest.mark.django_db embrulha cada teste numa transação que é revertida no final - nunca comita. E on_commit só dispara quando a transação comita de verdade. Ou seja: mesmo com a view perfeita, o callback nunca rodaria no teste. O teste ingênuo teria passado ou falhado por motivos que não têm nada a ver com o que ele diz verificar.

A ferramenta certa é uma fixture do próprio pytest-django:

with django_capture_on_commit_callbacks() as callbacks:
    resposta = client.post(reverse("checkout"), {...})
    assert resposta.status_code == 201
    assert len(callbacks) == 1

Sem execute=True: o teste prova a estrutura - exatamente um callback agendado. Nem zero (o bug clássico de enfileirar antes do commit), nem executado na hora (que seria não usar on_commit de jeito nenhum).

Testar com worker de verdade, sem abrir terminal

Os três gotchas acima se testam com o backend inline. Mas eu queria uma prova ponta a ponta: task enfileirada de verdade, worker real processando, falha real, e a rota HTTP devolvendo 422 - sem mock nenhum e sem depender de eu lembrar de subir um processo em outro terminal.

O db_worker tem --batch: processa o que está pronto e sai. Dá pra chamar de dentro do teste. A primeira tentativa quebrou:

OperationalError: cannot start a transaction within a transaction

O worker abre BEGIN EXCLUSIVE pra travar a fila com segurança contra outros workers. O SQLite não permite isso dentro de uma transação já aberta - e @pytest.mark.django_db abre uma. A correção é transaction=True, que desliga o embrulho e deixa o teste committar como produção comitaria:

@pytest.mark.django_db(transaction=True)
def test_laudo_invalido_falha_de_verdade_com_worker_real(client):
    ...
    resultado = resumir_laudo.enqueue(laudo.id)
    assert resultado.status == TaskResultStatus.READY
    call_command("db_worker", queue_name="laudos", batch=True, verbosity=0)
    resultado_final = default_task_backend.get_result(resultado.id)
    assert resultado_final.status == TaskResultStatus.FAILED
    assert resultado_final.errors[0].exception_class_path == "desk.domain.MicronaireForaDaFaixa"

É o teste mais lento da suíte, e é o único assim. O preço de verificar comportamento que só existe quando a transação comita mesmo.

A parede que não é do framework: mudar QUEUES não migra o que já está na fila

Essa não apareceu em teste. Apareceu rodando o projeto. Na fase inicial, a única fila era default. Depois migrei pra quatro filas nomeadas. Semanas de commits depois, subi um worker escutando tudo (--queue-name '*') e:

InvalidTaskError: Queue 'default' is not valid for backend.

Tasks enfileiradas semanas antes, com um nome de fila que não existia mais na configuração, continuavam paradas na tabela esperando alguém que nunca viria.

Mudar QUEUES altera o que o backend aceita daqui pra frente - não faz nada com o que já está gravado. Em ambiente de desenvolvimento, a limpeza é uma linha. Em produção, é um plano de migração: drenar a fila antiga antes do deploy, ou manter o nome antigo aceito durante a transição. É o tipo de coisa que não está em tutorial porque tutorial não tem histórico.

O painel, e a mentira que eu me recusei a contar

Construí um dashboard pra ver as filas trabalhando - cada fila é uma esteira, cada task uma etiqueta que muda de cor conforme o estado. E aí veio a frustração: as etiquetas nunca apareciam como "executando". Saltavam de READY direto pra SUCCESSFUL.

O motivo é honesto - resumir_laudo faz uma leitura de banco e uma formatação de string. Termina em milissegundos. O painel consulta a cada 1,5s. Não há o que ver.

A solução preguiçosa seria um time.sleep(2) dentro da task. Recusei: isso corromperia o código de negócio por efeito visual, e o painel passaria a mostrar uma latência que não existe.

Tentei antes o caminho legítimo: aumentar o --interval do worker. Ajudou pouco, e a descoberta foi interessante - o intervalo controla o tempo de espera quando a fila está vazia, não a duração da execução. Assim que o worker encontra sete laudos, processa os sete de enfiada.

A saída honesta foi separar as coisas: uma quinta fila, demo, com uma task cujo nome e docstring deixam explícito que a lentidão é a função dela.

@task(queue_name="demo", priority=0)
def tarefa_de_demonstracao() -> str:
    """
    Task artificialmente lenta, só para o painel exibir o estado 'executando'.
    O `sleep` aqui é intencional e honesto: simular latência É a função desta task.
    Ela não tem papel no negócio - existe apenas para tornar visível a transição
    READY → RUNNING → SUCCESSFUL, que nas tasks reais (rápidas) acontece rápido
    demais para o olho acompanhar.
    """
    time.sleep(4)

As quatro filas de negócio continuam rápidas e verdadeiras. A que mente sobre o tempo diz isso no próprio nome.

O que o framework não faz

Não há agendamento embutido. Não existe @task(run_every="0 18 * * *") - e a ingestão diária de preço precisa disso. A lacuna se preenche com um management command que o cron chama:

0 18 * * * cd /projeto && uv run python manage.py registrar_preco ICE-CT2 "$(preco.sh)" "$(date +\%Y-\%m-\%d)"

O padrão é "cron chama command, command enfileira task". Funciona, mas é peça a mais pra manter - e é a diferença mais concreta em relação ao Celery Beat.

Testes e o que ficou de fora

São 34 testes, TDD do primeiro ao último commit. A maioria roda inline com o ImmediateBackend, em ~1s. Um roda worker de verdade. Nenhum precisa de GOOGLE_API_KEY - a integração com PydanticAI (uma task que extrai dados estruturados de confirmações em texto livre) usa TestModel + Agent.override(), o mecanismo de teste da própria lib, sem tocar rede.

Ficou de fora, de propósito: Celery, WebSocket no painel, deploy, autenticação. Cada ausência está registrada num ADR.md com o motivo e o gatilho de revisão - a condição concreta que faria a decisão mudar. Trocar django-tasks-db por Celery agora, sem volume que justifique, seria otimização prematura; documentar quando trocar é mais útil que trocar. A escolha do polling de 1,5s em vez de SSE também virou ADR, com o trade-off vivido: transições mais curtas que o intervalo são invisíveis, e foi preciso uma task artificial pra enxergar RUNNING.

O padrão

Os três gotchas têm a mesma forma: o código de produção estava certo, e o teste falhou mesmo assim. Um backend que não implementa o que a API sugere. Um tipo que não sobre