Skip to Content
ReceitasPython (httpx)

Python (httpx + FastAPI)

pip install httpx fastapi uvicorn
zfiscoo_client.py
import httpx import os import hmac import hashlib ZF_API = os.environ["ZFISCOO_API_URL"] ZF_KEY = os.environ["ZFISCOO_API_KEY"] async def emitir_nfce(pedido_id: str, items: list, payment: dict) -> dict: async with httpx.AsyncClient(timeout=30.0) as client: r = await client.post( f"{ZF_API}/v1/nfce", headers={ "Authorization": f"Bearer {ZF_KEY}", "Idempotency-Key": f"nfce:{pedido_id}", }, json={ "issuer_id": os.environ["ZFISCOO_ISSUER_ID"], "external_ref": pedido_id, "items": items, "payment": payment, }, ) r.raise_for_status() return r.json() def verify_signature(raw_body: bytes, signature: str, secret: str) -> bool: expected = hmac.new(secret.encode(), raw_body, hashlib.sha256).hexdigest() received = signature.replace("sha256=", "") return hmac.compare_digest(expected, received)
main.py
from fastapi import FastAPI, Request, HTTPException from zfiscoo_client import emitir_nfce, verify_signature import os app = FastAPI() seen_events: set[str] = set() @app.post("/emitir") async def emitir(pedido: dict): return await emitir_nfce(pedido["id"], pedido["items"], pedido["payment"]) @app.post("/zfiscoo/webhook") async def webhook(req: Request): raw = await req.body() sig = req.headers.get("x-signature", "") if not verify_signature(raw, sig, os.environ["ZFISCOO_WEBHOOK_SECRET"]): raise HTTPException(401, "invalid signature") event = await req.json() if event["id"] in seen_events: return {"ok": True, "deduped": True} seen_events.add(event["id"]) print(f"[zfiscoo] {event['type']}{event['data']['id']}") return {"ok": True}
uvicorn main:app --port 8000