Receta de punto de venta (Python)

Esta receta implementa el flujo de la guía tirilla en un punto de venta: emitir con ?wait=N y una Idempotency-Key por venta, decidir qué imprimir con cada respuesta y armar la tirilla. Python 3.10 o superior, sin dependencias. El mismo archivo vive en ejemplos/python/ del repositorio.

Antes de empezar: pruébala con una clave de sandbox (sk_sandbox_...), que la persona escribe en .env. Nunca la pidas por el chat ni la imprimas.

Qué decide

sell(...) devuelve una de tres acciones:

`action`	Cuándo	Qué imprime el punto de venta
`tirilla`	El documento quedó `accepted` o en `contingency` y trae `print`.	La tirilla completa: `render(...)`.
`provisional`	Se acabó el plazo y el documento sigue en proceso, no hubo respuesta o `print` no llegó.	Un comprobante que dice que no es la factura: `provisional(...)`. La venta queda en tu cola.
`nada`	`rejected`, `failed` o una solicitud rechazada (`4xx`).	Nada que parezca una factura. Avisa a una persona.

Reglas que el código respeta:

La Idempotency-Key es la misma en todos los intentos de una venta, también tras un 202, un 429 o un corte de red.
Solo la primera solicitud espera (?wait=20). Las repeticiones van con ?wait=0 y después se consulta con GET.
Un 429 con limit: sync_open se repite de inmediato con ?wait=0; cualquier otro 429 o un 5xx espera Retry-After.
Un documento aceptado sin print se vuelve a leer; si no llega dentro del plazo, la venta queda provisional.

Paso 1. El archivo

"""Point of sale: issue one sale and decide what to print. Standard library only.

    result = sell(http(api_key), company_id, body, idempotency_key="venta-1042")
    if result.action == FULL:
        ticket = render(result.document, sale)      # the receipt, with its QR
    elif result.action == PROVISIONAL:
        slip = provisional("venta-1042", "caja@tienda.example")
        # ...and keep the sale in your own queue: call `sell` again later
        # with the SAME body and the SAME key.
    else:                                           # NOTHING: there is no invoice
        ...                                         # tell a person; print no receipt

One `Idempotency-Key` per sale, saved with the sale BEFORE the first request,
so a restart repeats it. Only the first request waits (`?wait=N`); every
repetition asks with `?wait=0` and the document is then read with GET.
"""

from __future__ import annotations

import json
import textwrap
import time
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from typing import Any, Callable

FULL = "tirilla"  # accepted or in contingency, with its `print` block
PROVISIONAL = "provisional"  # still in process, or its outcome is unknown: follow it up
NOTHING = "nada"  # no invoice: rejected, failed, or the request was refused

TITLES = {
    "invoice": "FACTURA ELECTRÓNICA DE VENTA",
    "credit_note": "NOTA CRÉDITO ELECTRÓNICA",
    "debit_note": "NOTA DÉBITO ELECTRÓNICA",
}

# (method, path, body, headers) -> (status, headers in lowercase, parsed body).
# It returns every HTTP status and raises OSError only when there is no answer.
Send = Callable[[str, str, Any, dict[str, str]], tuple[int, dict[str, str], Any]]


@dataclass
class Result:
    action: str
    document: dict[str, Any] | None  # None when no document is known
    reason: str  # a status of the document, an error code, or why it is provisional


@dataclass
class Ticket:
    head: list[str]
    qr: str  # the exact text the QR encodes: `print.qr_text`
    foot: list[str] = field(default_factory=list)


def http(
    api_key: str,
    base_url: str = "https://api.foliios.com",
    timeout: float = 30.0,  # must be longer than `wait`
    opener: Callable[..., Any] = urllib.request.urlopen,
) -> Send:
    def send(method: str, path: str, body: Any, headers: dict[str, str]) -> tuple[int, dict[str, str], Any]:
        data = None if body is None else json.dumps(body).encode()
        all_headers = {"Authorization": f"Bearer {api_key}", "Accept": "application/json", **headers}
        if data is not None:
            all_headers["Content-Type"] = "application/json"
        req = urllib.request.Request(base_url.rstrip("/") + path, data=data, method=method, headers=all_headers)
        try:
            with opener(req, timeout=timeout) as res:
                return res.status, _lower(res.headers), _parse(res.read())
        except urllib.error.HTTPError as err:  # an answer, not a network failure
            with err:
                return err.code, _lower(err.headers), _parse(err.read())

    return send


def sell(
    send: Send,
    company_id: str,
    body: dict[str, Any],
    idempotency_key: str,
    wait: int = 20,
    budget: float = 45.0,  # seconds the buyer stands in front of the machine
    sleep: Callable[[float], None] = time.sleep,
    clock: Callable[[], float] = time.monotonic,
) -> Result:
    deadline = clock() + budget
    path = f"/v1/companies/{company_id}/documents"
    document: dict[str, Any] | None = None
    seconds = wait
    while True:
        try:
            if document is None:
                # Always the same key: a `202`, a `429` or a cut connection never gets a new one.
                status, headers, payload = send("POST", f"{path}?wait={seconds}", body, {"Idempotency-Key": idempotency_key})
            else:
                status, headers, payload = send("GET", f"{path}/{document['id']}", None, {})
        except OSError:  # no answer: the request may have arrived
            status, headers, payload = None, {}, None
        waited, seconds = seconds > 0, 0  # a repetition does not wait, so it takes no `sync_open` slot
        problem = payload if isinstance(payload, dict) else {}
        code = problem.get("code")
        pause = _retry_after(headers, 2.0)
        if status in (200, 201, 202):
            document = payload
            decided = decide(document)
            if decided:
                return decided
        elif status == 429 and problem.get("limit") == "sync_open" and waited:
            pause = 0.0  # every waiting slot is taken and nothing was created: ask again at once, without waiting
        elif status is None or status == 429 or status >= 500 or code == "concurrent_modification":
            pass  # retry after the pause, same key and same body
        elif document is None:
            return Result(NOTHING, None, code or f"http_{status}")  # refused: no number, no document
        else:
            return Result(PROVISIONAL, document, code or f"http_{status}")  # it exists but cannot be read now
        if clock() + pause > deadline:
            return Result(PROVISIONAL, document, _pending_reason(document))
        sleep(pause)


def decide(document: dict[str, Any]) -> Result | None:
    """What to print for a document, or None while it has to be read again."""
    status = document.get("status")
    if status in ("accepted", "contingency"):
        # Without `print` there is no QR, legend nor software line: read again.
        return Result(FULL, document, status) if document.get("print") else None
    if status in ("rejected", "failed"):
        return Result(NOTHING, document, status)
    return None  # pending, sending, dian_unavailable (also with `dian_outcome_uncertain`)


def render(document: dict[str, Any], sale: dict[str, Any], width: int = 42) -> Ticket:
    """The receipt of an `accepted` or `contingency` document.

    `sale` is what the API does not return and the point of sale already has:
    `issuer` (`name`, `nit`, `responsibilities`, `address`), `lines`
    (`description`, `quantity`, `total`), `taxes` (`name`, `base`, `amount`)
    and `payment` (form and means, in words). Amounts are texts already
    formatted: nothing is computed here.
    """
    block = document["print"]
    issuer = sale["issuer"]
    customer = document["customer"]
    totals = document["totals"]
    title = TITLES.get(document["type"], TITLES["invoice"])
    if document["status"] == "contingency":
        title += " (CONTINGENCIA DIAN)"

    head = [line.center(width) for line in _wrap(title, width)]
    head += [line.center(width) for line in _wrap(issuer["name"], width)]
    head.append(f"NIT {issuer['nit']}".center(width))
    for extra in (issuer.get("responsibilities"), issuer.get("address")):
        if extra:
            head += [line.center(width) for line in _wrap(extra, width)]
    head.append("-" * width)
    head.append(_pair("No.", document["number"], width))
    head.append(_pair("Generación", document["issued_at"].replace("T", " ")[:19], width))
    identification = (customer.get("identification") or {}).get("number", "")
    head += _wrap(f"Adquirente: {customer['name']} {identification}".strip(), width)
    head.append("-" * width)
    for line in sale["lines"]:
        head += _wrap(f"{line['quantity']} x {line['description']}", width)
        head.append(line["total"].rjust(width))
    head.append("-" * width)
    for tax in sale.get("taxes", []):
        head.append(_pair(f"{tax['name']} base {tax['base']}", tax["amount"], width))
    head.append(_pair("Subtotal", totals["tax_exclusive_amount"], width))
    head.append(_pair(f"TOTAL {document['currency']}", totals["payable_amount"], width))
    head += _wrap(f"Pago: {sale['payment']}", width)

    foot = _wrap(block["status_label"], width)
    foot += _wrap(f"{'CUFE' if document['type'] == 'invoice' else 'CUDE'}: {block['cufe']}", width)
    foot += _wrap(f"Consulte en {block['qr_url'].split('/')[2]}", width)
    numbering = block.get("numbering_authorization")
    if numbering:
        series = f"prefijo {numbering['prefix']} " if numbering["prefix"] else ""
        foot += _wrap(
            f"Autorización de numeración DIAN {numbering['number']}, {series}"
            f"del {numbering['from']} al {numbering['to']}, "
            f"vigente del {numbering['valid_from']} al {numbering['valid_to']}",
            width,
        )
    software = block["software"]
    made_by = f"Software: {software['name']} · Fabricante: {software['manufacturer_name']}"
    if software.get("manufacturer_id"):  # `null` when the document declares none
        made_by += f", NIT {software['manufacturer_id']}"
    foot += _wrap(made_by, width)
    if document.get("environment") != "live":
        foot.append(f"SIN VALIDEZ FISCAL ({document.get('environment')})".center(width))
    return Ticket(head, block["qr_text"], foot)


def provisional(sale_reference: str, contact: str, width: int = 42) -> list[str]:
    """The slip for a sale whose invoice is not ready: it is not the invoice."""
    lines = ["COMPROBANTE PROVISIONAL".center(width)]
    lines += _wrap("Este comprobante no es la factura electrónica de venta.", width)
    lines += _wrap("Su factura está en proceso de validación ante la DIAN.", width)
    lines += _wrap(f"Referencia de la venta: {sale_reference}", width)
    lines += _wrap(f"Solicítela con esta referencia en {contact}", width)
    return lines


def _pending_reason(document: dict[str, Any] | None) -> str:
    if document is None:
        return "unknown"  # no answer ever arrived: the document may or may not exist
    if document.get("status") in ("accepted", "contingency"):
        return "print_unavailable"
    return document.get("outcome_code") or document.get("status") or "unknown"


def _wrap(text: str, width: int) -> list[str]:
    return textwrap.wrap(text, width, break_on_hyphens=False) or [""]


def _pair(left: str, right: str, width: int) -> str:
    return left + right.rjust(max(width - len(left), len(right) + 1))


def _retry_after(headers: dict[str, str], default: float) -> float:
    try:
        return max(float(headers["retry-after"]), 0.0)
    except (KeyError, ValueError):
        return default


def _lower(headers: Any) -> dict[str, str]:
    return {k.lower(): v for k, v in headers.items()} if headers else {}


def _parse(raw: bytes) -> Any:
    try:
        return json.loads(raw) if raw else None
    except ValueError:
        return None

render no calcula montos: recibe los de tu venta ya formateados y toma de la respuesta los totales, el CUFE, el texto del QR, la leyenda, la autorización de numeración y el software. Devuelve el texto de la tirilla en dos partes, antes y después del QR, y el texto exacto que el QR debe codificar (print.qr_text). Dibujar el QR y enviar los bytes a la impresora depende de tu equipo.

Paso 2. Probar sin red

import unittest

import tirilla

COMPANY = "cmp_unit"
KEY = "venta-1042"
BODY = {"resolution_id": "18760000001", "type": "invoice", "lines": [{"description": "Cuaderno argollado", "quantity": "1", "unit_price": "4000.00"}]}
CODE = "9f3c" * 24  # 96 characters, as a CUFE has
URL = f"https://catalogo-vpfe.dian.gov.co/document/searchqr?documentkey={CODE}"

SALE = {
    "issuer": {"name": "Tienda de Ejemplo S.A.S.", "nit": "900000001-9", "responsibilities": "Responsable de IVA", "address": "Calle 1 # 2-3, Ciudad Ejemplo"},
    "lines": [{"description": "Cuaderno argollado", "quantity": "1", "total": "4000.00"}],
    "taxes": [{"name": "IVA 19%", "base": "4000.00", "amount": "760.00"}],
    "payment": "Contado · Efectivo",
}


def document(status: str, with_print: bool = True, **extra) -> dict:
    validated = status == "accepted"
    block = {
        "cufe": CODE,
        "qr_text": f"NumFac: KA1042\nCUFE: {CODE}\n{URL}",
        "qr_url": URL,
        "validated_at": "2026-11-11T14:32:09-05:00" if validated else None,
        "status_label": "Documento validado por la DIAN el 11/11/2026 a las 14:32:09" if validated else "Expedida en contingencia DIAN, sin validación previa",
        "software": {"name": "Caja Ejemplo", "manufacturer_name": "Software de Ejemplo S.A.S.", "manufacturer_id": "900000002-7", "software_id": "56f2ae4e-9812-4fad-9255-08fcfcd5ccb0"},
        "numbering_authorization": {"number": "18760000001", "prefix": "KA", "from": 1, "to": 50000, "valid_from": "2026-01-01", "valid_to": "2027-12-31"},
    }
    return {
        "id": "doc_1",
        "environment": "live",
        "type": "invoice",
        "number": "KA1042",
        "status": status,
        "cufe": CODE if status in ("accepted", "contingency") else None,
        "issued_at": "2026-11-11T14:32:07-05:00",
        "print": block if with_print and status in ("accepted", "contingency") else None,
        "customer": {"name": "consumidor final", "identification": {"type": "citizenship_card", "number": "2" * 12}},
        "currency": "COP",
        "totals": {"tax_exclusive_amount": "4000.00", "payable_amount": "4760.00"},
        **extra,
    }


class Script:
    """A scripted API: each call takes the next answer; `OSError` is a cut connection."""

    def __init__(self, *answers):
        self.answers = list(answers)
        self.calls: list[tuple[str, str, dict]] = []
        self.now = 0.0
        self.sleeps: list[float] = []

    def send(self, method, path, body, headers):
        self.calls.append((method, path, headers))
        answer = self.answers.pop(0) if len(self.answers) > 1 else self.answers[0]
        if answer is OSError:
            raise OSError("connection reset")
        status, res_headers, payload = answer
        return status, res_headers, payload

    def sleep(self, seconds):
        self.sleeps.append(seconds)
        self.now += seconds

    def clock(self):
        return self.now

    def sell(self, **options):
        return tirilla.sell(self.send, COMPANY, BODY, KEY, sleep=self.sleep, clock=self.clock, **options)

    def posts(self):
        return [call for call in self.calls if call[0] == "POST"]


class SellTest(unittest.TestCase):
    def test_an_accepted_sale_prints_the_receipt_with_one_request(self):
        api = Script((201, {}, document("accepted")))
        result = api.sell()
        self.assertEqual((result.action, result.reason), (tirilla.FULL, "accepted"))
        self.assertEqual(api.calls, [("POST", f"/v1/companies/{COMPANY}/documents?wait=20", {"Idempotency-Key": KEY})])

    def test_a_contingency_prints_the_receipt_too(self):
        result = Script((201, {}, document("contingency"))).sell()
        self.assertEqual((result.action, result.reason), (tirilla.FULL, "contingency"))

    def test_a_rejected_or_failed_document_prints_nothing(self):
        for status in ("rejected", "failed"):
            result = Script((200, {}, document(status))).sell()
            self.assertEqual((result.action, result.reason), (tirilla.NOTHING, status))
            self.assertEqual(result.document["id"], "doc_1")

    def test_a_refused_request_prints_nothing_and_has_no_document(self):
        api = Script((422, {}, {"code": "validation_failed", "errors": []}))
        result = api.sell()
        self.assertEqual((result.action, result.document, result.reason), (tirilla.NOTHING, None, "validation_failed"))
        self.assertEqual(len(api.calls), 1)

    def test_a_202_is_followed_with_get_and_never_issued_again(self):
        api = Script((202, {"retry-after": "2"}, document("sending")), (200, {}, document("sending")), (200, {}, document("accepted")))
        result = api.sell()
        self.assertEqual(result.action, tirilla.FULL)
        self.assertEqual([call[:2] for call in api.calls[1:]], [("GET", f"/v1/companies/{COMPANY}/documents/doc_1")] * 2)
        self.assertEqual(len(api.posts()), 1)
        self.assertEqual(api.sleeps, [2.0, 2.0])

    def test_an_uncertain_outcome_ends_provisional_without_another_key(self):
        uncertain = document("sending", outcome_code="dian_outcome_uncertain")
        api = Script((202, {"retry-after": "2"}, uncertain), (200, {}, uncertain))
        result = api.sell(budget=10)
        self.assertEqual((result.action, result.reason), (tirilla.PROVISIONAL, "dian_outcome_uncertain"))
        self.assertEqual(result.document["id"], "doc_1")
        self.assertEqual(len(api.posts()), 1)
        self.assertLessEqual(api.now, 10)

    def test_a_document_that_fails_after_the_202_prints_nothing(self):
        api = Script((202, {}, document("pending")), (200, {}, document("rejected")))
        self.assertEqual(api.sell().action, tirilla.NOTHING)

    def test_a_full_sync_open_quota_asks_again_at_once_without_waiting(self):
        full = (429, {"retry-after": "25"}, {"code": "concurrency_limited", "limit": "sync_open"})
        api = Script(full, (202, {}, document("pending")), (200, {}, document("accepted")))
        result = api.sell()
        self.assertEqual(result.action, tirilla.FULL)
        self.assertEqual([call[1].rsplit("?", 1)[1] for call in api.posts()], ["wait=20", "wait=0"])
        self.assertEqual({call[2]["Idempotency-Key"] for call in api.posts()}, {KEY})
        self.assertEqual(api.sleeps[0], 0.0)

    def test_another_429_waits_its_retry_after_and_keeps_the_key(self):
        api = Script((429, {"retry-after": "3"}, {"code": "rate_limited", "limit": "writes"}), (201, {}, document("accepted")))
        self.assertEqual(api.sell().action, tirilla.FULL)
        self.assertEqual(api.sleeps, [3.0])
        self.assertEqual({call[2]["Idempotency-Key"] for call in api.posts()}, {KEY})

    def test_a_429_longer_than_the_budget_ends_provisional(self):
        api = Script((429, {"retry-after": "60"}, {"code": "rate_limited", "limit": "writes"}))
        result = api.sell()
        self.assertEqual((result.action, result.document, result.reason), (tirilla.PROVISIONAL, None, "unknown"))
        self.assertEqual(api.sleeps, [])

    def test_a_cut_connection_repeats_the_same_key_without_waiting(self):
        api = Script(OSError, (503, {"retry-after": "1"}, {"code": "temporarily_unavailable"}), (201, {}, document("accepted")))
        result = api.sell()
        self.assertEqual(result.action, tirilla.FULL)
        self.assertEqual([call[1].rsplit("?", 1)[1] for call in api.posts()], ["wait=20", "wait=0", "wait=0"])
        self.assertEqual({call[2]["Idempotency-Key"] for call in api.posts()}, {KEY})

    def test_no_answer_at_all_ends_provisional(self):
        api = Script(OSError)
        result = api.sell(budget=6)
        self.assertEqual((result.action, result.document, result.reason), (tirilla.PROVISIONAL, None, "unknown"))
        self.assertTrue(all(call[2] == {"Idempotency-Key": KEY} for call in api.calls))

    def test_an_accepted_document_without_print_is_read_again(self):
        api = Script((201, {}, document("accepted", with_print=False)), (200, {}, document("accepted")))
        self.assertEqual(api.sell().action, tirilla.FULL)
        still = Script((201, {}, document("accepted", with_print=False)))
        result = still.sell(budget=6)
        self.assertEqual((result.action, result.reason), (tirilla.PROVISIONAL, "print_unavailable"))


class RenderTest(unittest.TestCase):
    def lines(self, ticket):
        return ticket.head + ticket.foot

    def test_the_receipt_fits_the_paper_and_carries_what_print_brings(self):
        for width in (32, 42):
            ticket = tirilla.render(document("accepted"), SALE, width)
            lines = self.lines(ticket)
            self.assertTrue(all(len(line) <= width for line in lines), [line for line in lines if len(line) > width])
            text = "".join(lines)
            self.assertIn(CODE, text.replace("CUFE: ", ""))
            self.assertEqual(ticket.qr, document("accepted")["print"]["qr_text"])
            flat = " ".join(line.strip() for line in lines)
            for expected in (
                "FACTURA ELECTRÓNICA DE VENTA",
                "NIT 900000001-9",
                "KA1042",
                "2026-11-11 14:32:07",
                "consumidor final",
                "4760.00",
                "Contado · Efectivo",
                "Documento validado por la DIAN el 11/11/2026 a las 14:32:09",
                "catalogo-vpfe.dian.gov.co",
                "Autorización de numeración DIAN 18760000001, prefijo KA del 1 al 50000, vigente del 2026-01-01 al 2027-12-31",
                "Software: Caja Ejemplo · Fabricante: Software de Ejemplo S.A.S., NIT 900000002-7",
            ):
                self.assertIn(expected, flat)
            self.assertNotIn("SIN VALIDEZ FISCAL", flat)

    def test_a_contingency_says_so_and_shows_no_validation(self):
        flat = " ".join(line.strip() for line in self.lines(tirilla.render(document("contingency"), SALE)))
        self.assertIn("(CONTINGENCIA DIAN)", flat)
        self.assertIn("Expedida en contingencia DIAN, sin validación previa", flat)
        self.assertNotIn("validado por la DIAN", flat)

    def test_a_manufacturer_without_identification_prints_no_nit(self):
        doc = document("accepted")
        doc["print"]["software"]["manufacturer_id"] = None
        flat = " ".join(line.strip() for line in self.lines(tirilla.render(doc, SALE)))
        self.assertIn("Fabricante: Software de Ejemplo S.A.S.", flat)
        self.assertNotIn("S.A.S., NIT", flat)

    def test_a_document_that_is_not_live_says_it_has_no_fiscal_validity(self):
        flat = " ".join(self.lines(tirilla.render(document("accepted", environment="sandbox"), SALE)))
        self.assertIn("SIN VALIDEZ FISCAL (sandbox)", flat)

    def test_the_provisional_slip_says_it_is_not_the_invoice(self):
        slip = tirilla.provisional(KEY, "caja@tienda.example", 32)
        self.assertTrue(all(len(line) <= 32 for line in slip))
        flat = " ".join(line.strip() for line in slip)
        self.assertIn("no es la factura electrónica de venta", flat)
        self.assertIn(KEY, flat)
        self.assertNotIn("CUFE", flat)


if __name__ == "__main__":
    unittest.main()

python3 -I -m unittest

Deben pasar todas. No llaman a la API ni leen la clave.

Paso 3. Conectarlo a tu punto de venta

Guarda la venta con su clave antes de llamar a sell. La clave sale de la venta (por ejemplo, su número interno).
Arma el cuerpo como en emitir una factura. Para consumidor final, omite customer.
Con tirilla: imprime head, el QR de qr y foot, y guarda el id del documento para reimprimir.
Con provisional: imprime el comprobante y deja la venta en una cola. Un proceso de fondo llama otra vez a sell con el mismo cuerpo y la misma clave hasta que la acción sea tirilla o nada.
Con nada: registra reason y avisa a una persona. No reintentes sin corregir.

En el sandbox puedes forzar cada rama agregando en tu función send la cabecera Foliios-Sandbox-Scenario (escenarios): un rechazo, una respuesta lenta o la contingencia (dian_unavailable).

Qué no hace

No dibuja el QR ni habla con la impresora.
No guarda la cola de ventas provisionales: es de tu sistema.
No decide qué entregar al comprador cuando no hay factura. Ver la guía.