Skip to content

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


Terminal window
pip install httpx

Variables de entorno en .env (con python-decouple o equivalente):

EMITIA_API_KEY=sk_test_TU_CLAVE
EMITIA_WEBHOOK_SECRET=whsec_...

emitia/client.py
from __future__ import annotations
import hashlib
import hmac
import time
import uuid
from 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()

emitia/services.py
from __future__ import annotations
import os
import 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()

emitia/views.py
from __future__ import annotations
import hashlib
import hmac
import json
import logging
import os
import time
from django.http import HttpRequest, HttpResponse, JsonResponse
from django.views.decorators.csrf import csrf_exempt
from 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_POST
def 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})

urls.py
from django.urls import path
from emitia.views import emitia_webhook
urlpatterns = [
path("webhooks/emitia/", emitia_webhook),
]