Síncrono, assíncrono e polling

Este guia trata de duas coisas diferentes que costumam aparecer juntas: o cliente assíncrono do Python (AsyncNfeClient, com async/await) e a emissão assíncrona da API, em que a nota leva alguns segundos para ser autorizada pela prefeitura.

A emissão é assíncrona

service_invoices.create responde quase sempre com 202 Accepted. A nota volta com id e com um flow_status intermediário (WaitingCalculateTaxes ou WaitingSend, por exemplo) e segue sendo processada até um estado final, em geral em poucos segundos.

Os estados conhecidos estão em nfeio.types.FlowStatus:

Grupo	Valores
Sucesso	Issued, Cancelled (FlowStatus.SUCCESS)
Falha	IssueFailed, CancelFailed, Error (FlowStatus.FAILURE)
Em andamento	WaitingCalculateTaxes, WaitingDefineRpsNumber, WaitingSend, WaitingReturn, WaitingSendCancel, WaitingReturnCancel, WaitingDownload, PullFromCityHall

O conjunto é aberto: compare strings. Um valor que o SDK não conhece é tratado como "ainda em andamento".

Três formas de saber o resultado

Forma	Quando usar
create_and_wait	Scripts, jobs e filas: envia e espera no mesmo passo.
create + wait	Quando você precisa gravar o id antes de esperar.
create + webhook	Aplicações web: responda ao usuário na hora e receba o resultado por push.

create_and_wait

from decimal import Decimal

from nfeio import InvoiceProcessingError, NfeClient, PollingTimeoutError

client = NfeClient()
company_id = "ID_DA_EMPRESA"

try:
    invoice = client.service_invoices.create_and_wait(
        company_id,
        {
            "cityServiceCode": "10677",
            "description": "Consultoria em tecnologia",
            "servicesAmount": Decimal("150.00"),
        },
        external_id="pedido-1002",
        timeout=180,  # segundos de espera; o padrão é 120
    )
    print(invoice.flow_status)  # "Issued"
except InvoiceProcessingError as erro:
    # A prefeitura recusou: IssueFailed ou Error. flow_message explica o motivo.
    print(erro.flow_status, erro.flow_message)
except PollingTimeoutError as erro:
    # O prazo acabou antes de um estado final. A nota existe e continua sendo processada.
    print("ainda processando:", erro.flow_status, erro.invoice.id if erro.invoice else None)

PollingTimeoutError não significa que a emissão falhou: a nota foi aceita e continua na fila da prefeitura. Guarde o id e consulte depois com retrieve ou wait. Nunca emita de novo por causa de um timeout de espera.

create + wait

from decimal import Decimal

from nfeio import NfeClient

client = NfeClient()
company_id = "ID_DA_EMPRESA"

accepted = client.service_invoices.create(
    company_id,
    {"cityServiceCode": "10677", "description": "Suporte", "servicesAmount": Decimal("80.00")},
    external_id="pedido-1003",
)
assert accepted.id is not None
print(accepted.id, accepted.flow_status)  # id já disponível, ex.: WaitingCalculateTaxes

invoice = client.service_invoices.wait(company_id, accepted.id)
print(invoice.flow_status)

Parâmetros de espera

wait, create_and_wait e cancel_and_wait aceitam os mesmos parâmetros:

Parâmetro	Padrão	O que faz
timeout	120.0	Tempo máximo de espera, em segundos.
initial_interval	1.0	Intervalo da primeira consulta.
max_interval	10.0	Teto do intervalo entre consultas.
backoff	1.5	Multiplicador do intervalo a cada consulta (entre 1 e 10).
not_found_grace	10.0	Por quanto tempo um 404 logo após o 202 é tolerado.
raise_on_failure	True	Com False, um estado de falha é devolvido em vez de levantar InvoiceProcessingError.

O polling usa GET da própria nota (a API não tem rota de status), com backoff até max_interval.

Cancelamento

O cancelamento também é assíncrono. cancel_and_wait pede o cancelamento e espera Cancelled:

from nfeio import NfeClient

client = NfeClient()

invoice = client.service_invoices.cancel_and_wait("ID_DA_EMPRESA", "ID_DA_NOTA")
assert invoice.flow_status == "Cancelled"

Se a prefeitura recusar, sobe InvoiceProcessingError com flow_status == "CancelFailed".

Cliente assíncrono

AsyncNfeClient tem os mesmos recursos, métodos, parâmetros e comportamento do NfeClient. A diferença é que cada método é uma corrotina, e a paginação usa async for.

import asyncio
from decimal import Decimal

from nfeio import AsyncNfeClient


async def main() -> None:
    async with AsyncNfeClient() as client:
        invoice = await client.service_invoices.create_and_wait(
            "ID_DA_EMPRESA",
            {
                "cityServiceCode": "10677",
                "description": "Consultoria em tecnologia",
                "servicesAmount": Decimal("150.00"),
            },
            external_id="pedido-1004",
        )
        print(invoice.flow_status)

        page = await client.companies.list()
        async for company in page.auto_paging_iter():
            print(company.id, company["name"])


asyncio.run(main())

Para emitir várias notas em paralelo, use asyncio.gather com um external_id por nota e return_exceptions=True, para que a falha de uma nota não esconda o resultado das outras:

import asyncio
from decimal import Decimal

from nfeio import AsyncNfeClient, ServiceInvoice


async def emitir_lote(
    pedidos: list[tuple[str, Decimal]],
) -> list[ServiceInvoice | BaseException]:
    async with AsyncNfeClient() as client:
        tarefas = [
            client.service_invoices.create_and_wait(
                "ID_DA_EMPRESA",
                {"cityServiceCode": "10677", "description": "Assinatura", "servicesAmount": valor},
                external_id=f"pedido-{pedido_id}",
            )
            for pedido_id, valor in pedidos
        ]
        return await asyncio.gather(*tarefas, return_exceptions=True)


resultados = asyncio.run(emitir_lote([("2001", Decimal("49.90")), ("2002", Decimal("99.90"))]))
for resultado in resultados:
    if isinstance(resultado, BaseException):
        print("falhou:", repr(resultado))  # trate como em "Tratamento de erros"
    else:
        print(resultado.id, resultado.flow_status)

Como funciona por dentro: o transporte padrão roda a E/S em threads (asyncio.to_thread) e as esperas usam asyncio.sleep, então o event loop nunca é bloqueado. Para uma pilha HTTP nativamente assíncrona, implemente nfeio.transport.AsyncTransport.

Cancelar uma task não cancela a requisição

Cancelar a task não interrompe uma requisição que já está em voo: ela termina na thread. Se a task cancelada estava emitindo uma nota, trate como resultado incerto e reconcilie por external_id antes de tentar de novo (veja Tratamento de erros).

Veja também

• Tratamento de erros: outcome_unknown e a reconciliação.
• Webhooks: o resultado por push.
• Notas de serviço: todos os métodos.