Django (Python)
Integración con Django usando httpx para llamadas síncronas.
El webhook usa Django’s csrf_exempt y verifica la firma HMAC.
Emitia-Version: 2026-01-01
Instalación
Section titled “Instalación”pip install httpxVariables de entorno en .env (con python-decouple o equivalente):
EMITIA_API_KEY=sk_test_TU_CLAVEEMITIA_WEBHOOK_SECRET=whsec_...Cliente Emitia
Section titled “Cliente Emitia”from __future__ import annotations
import hashlibimport hmacimport timeimport uuidfrom typing import Any
import httpx
API_BASE = "https://api.emitia.co/v1"EMITIA_VERSION = "2026-01-01"
class EmitiaClient: def __init__(self, api_key: str) -> None: self._client = httpx.Client( base_url=API_BASE, headers={ "Authorization": f"Bearer {api_key}", "Emitia-Version": EMITIA_VERSION, }, timeout=30, )
def get_account(self) -> dict[str, Any]: """Verifica autenticación (funciona hoy).""" return self._client.get("/account").raise_for_status().json()
def create_customer(self, data: dict[str, Any]) -> dict[str, Any]: """Crea un cliente. Disponible con api-resources.""" idempotency_key = str(uuid.uuid4()) return ( self._client.post( "/customers", json=data, headers={"Idempotency-Key": idempotency_key}, ) .raise_for_status() .json() )
def create_invoice(self, data: dict[str, Any], idempotency_key: str) -> dict[str, Any]: """ Crea una factura. Disponible con api-resources + test-mode-sandbox. El caller genera y persiste idempotency_key ANTES de llamar. """ return ( self._client.post( "/invoices", json=data, headers={"Idempotency-Key": idempotency_key}, ) .raise_for_status() .json() )
def close(self) -> None: self._client.close()Crear cliente y factura
Section titled “Crear cliente y factura”from __future__ import annotations
import osimport uuid
from .client import EmitiaClient
def create_invoice_for_order( customer_nit: str, customer_name: str, description: str, unit_price: str, order_id: str,) -> dict: client = EmitiaClient(api_key=os.environ["EMITIA_API_KEY"]) try: # 1. Crear cliente customer = client.create_customer({ "organization_type": "1", "identification": { "type": "31", "number": customer_nit, "check_digit": "1", }, "legal_name": customer_name, "email": "facturacion@cliente.co", })
# 2. Generar clave de idempotencia y persistirla ANTES de llamar idempotency_key = str(uuid.uuid4()) # Aquí deberías guardar idempotency_key en tu DB asociado al pedido
tax_value = f"{float(unit_price) * 0.19:.2f}"
# 3. Crear factura invoice = client.create_invoice( data={ "customer_id": customer["id"], "currency": "COP", "operation_type": "10", "payment_means": {"code": "10", "type": "1"}, "lines": [ { "description": description, "quantity": "1.00", "unit_code": "94", "unit_price": unit_price, "line_extension_amount": unit_price, "taxes": [ { "code": "01", "rate": "19.00", "base": unit_price, "value": tax_value, } ], } ], "metadata": {"order_id": order_id}, }, idempotency_key=idempotency_key, ) return invoice finally: client.close()Vista de webhook
Section titled “Vista de webhook”from __future__ import annotations
import hashlibimport hmacimport jsonimport loggingimport osimport time
from django.http import HttpRequest, HttpResponse, JsonResponsefrom django.views.decorators.csrf import csrf_exemptfrom django.views.decorators.http import require_POST
logger = logging.getLogger(__name__)
def _verify_signature(raw_body: bytes, signature: str, secret: str, tolerance_sec: int = 300) -> bool: """Verifica la firma HMAC del webhook de Emitia.""" parts = dict(item.split("=", 1) for item in signature.split(",")) t_str = parts.get("t", "") v1 = parts.get("v1", "") if not t_str or not v1: return False t = int(t_str) if abs(time.time() - t) > tolerance_sec: return False payload = f"{t}.{raw_body.decode('utf-8')}" expected = hmac.new( secret.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256, ).hexdigest() return hmac.compare_digest(expected, v1)
@csrf_exempt@require_POSTdef emitia_webhook(request: HttpRequest) -> HttpResponse: raw_body = request.body signature = request.headers.get("Emitia-Signature", "") secret = os.environ.get("EMITIA_WEBHOOK_SECRET", "")
if not _verify_signature(raw_body, signature, secret): return JsonResponse({"error": "Firma inválida"}, status=401)
event = json.loads(raw_body) event_type = event.get("type", "")
if event_type == "invoice.accepted_by_dian": invoice = event["data"]["object"] logger.info("Factura aceptada: %s CUFE: %s", invoice["id"], invoice.get("cufe", "")) # Actualizar tu base de datos aquí elif event_type == "invoice.rejected_by_dian": invoice = event["data"]["object"] logger.error("Factura rechazada: %s", invoice["id"])
return JsonResponse({"received": True})URL configuration
Section titled “URL configuration”from django.urls import pathfrom emitia.views import emitia_webhook
urlpatterns = [ path("webhooks/emitia/", emitia_webhook),]Qué sigue
Section titled “Qué sigue”- Conceptos: Webhooks — detalles de la verificación HMAC
- Conceptos: Idempotencia — el patrón de retry seguro
- Guía de mapeo POS — códigos DIAN para tu sistema