wsbuilder 0.6.0

Framework ligero HTTP + WebSocket.

wsbuilder

wsbuilder es un paquete Python para construir servidores HTTP + WebSocket sin dependencias externas.

Lightweight Python HTTP + WebSocket framework for building real-time APIs and custom servers.

Keywords: python, http-server, websocket, framework, real-time, api, socket.

Incluye

• wsbuilder.framework: router, request/response, servidor HTTP, handshake WS y utilidades de frames.
• wsbuilder.orm: ORM flexible para SQLite3 (modelos, QuerySet, filtros, transacciones anidadas).
• wsbuilder.dns: DNS UDP local minimo para registros localhost.
• wsbuilder.cookies / wsbuilder.headers: utilidades para cookies y headers HTTP.
• wsbuilder.ws_demo: demo completo HTTP + WebSocket + REST + SQLite.

Documentacion completa

• Ver docs/README.md para arquitectura, API, seguridad, CI/CD, testing y guias de uso.

Instalacion local

python -m pip install -e .

Uso rapido del framework

from wsbuilder import App, Response, parse_close_payload

app = App()

@app.view("/")
def home(_request):
    return Response.html("<h1>wsbuilder</h1>")

@app.api("/api/health")
def health(_request):
    return {"ok": True}

@app.ws("/ws/")
def ws_handler(ws, _request):
    while True:
        fin, opcode, payload, _masked, _mask = ws.recv_frame()
        if opcode == 0x8:
            code, reason = parse_close_payload(payload)
            ws.close(code or 1000, reason or "")
            break
        if opcode == 0x9:
            ws.send_pong(payload)
            continue
        if opcode == 0x1:
            ws.send_text(payload.decode("utf-8", errors="ignore"))
        elif opcode == 0x2:
            ws.send_binary(payload)

app.run("0.0.0.0", 8765)

Para CORS sin variables de entorno:

app = App(cors_allow_origin="*")

Thread Pool en Views

Las view por defecto se ejecutan en el hilo padre (thread_count=0). Puedes configurar un pool por ruta con min/max de hilos y limite de requests por hilo:

from wsbuilder import App

app = App()

@app.view("/jobs", min_threads=1, max_threads=4, requests_per_thread=0)
def jobs(_request):
    return "ok"

• min_threads: workers iniciales para la view.
• max_threads: tope de workers para esa view (0 deshabilita pool).
• requests_per_thread: capacidad por worker (0 = sin limite).
• Distribucion: siempre por least_busy (worker menos ocupado).
• Se expone metadata de worker en headers y metrics (/api/metrics, /api/metrics/stream).

ORM SQLite3 rapido

from wsbuilder import Database, IntegerField, Model, TextField

class User(Model):
    id = IntegerField(primary_key=True, auto_increment=True)
    username = TextField(unique=True, index=True, null=False)
    email = TextField(null=False)

db = Database("app.db")
User.create_table(db)

u = User(username="alice", email="[e-mail]")
u.save(db)

admins = User.objects(db).filter(username__contains="ali").order_by("-id").all()
print([x.to_dict() for x in admins])

Metrics (JSON stream)

from wsbuilder import App

app = App()
app.enable_metrics()  # crea /api/metrics y /api/metrics/stream

Probar snapshot:

curl http://127.0.0.1:8765/api/metrics

Probar stream NDJSON:

curl -N "http://127.0.0.1:8765/api/metrics/stream?interval=1&limit=5"

Modo continuo (no termina solo):

curl -N "http://127.0.0.1:8765/api/metrics/stream?interval=1&follow=1"

DNS local minimo

from wsbuilder import LocalDNSServer

dns = LocalDNSServer()  # default: 127.0.0.1:5533
dns.start()
# ...
dns.close()

HTTP streaming (chunked)

import time
from wsbuilder import App, Response

app = App()

@app.view("/stream")
def stream(_request):
    def chunks():
        for i in range(5):
            yield f"chunk {i}\n"
            time.sleep(1)
    return Response.stream(chunks(), content_type="text/plain; charset=utf-8")

Ejecutar demo incluido

python -m wsbuilder --host 0.0.0.0 --port 8765
# o
wsbuilder --host 0.0.0.0 --port 8765

Configuracion CORS

• Usa App(cors_allow_origin="https://tu-dominio.com").
• Usa App(cors_allow_origin="*") para permitir cualquier origen.

CI/CD (GitHub Actions)

• package-build.yml: construye sdist + wheel, valida metadata y prueba instalacion/import.
• release-from-main.yml en push a main: calcula semver automaticamente, crea rama release/v<version> desde main, actualiza version del paquete, crea tag v<version> y publica GitHub Release.
• publish-packages.yml en push de tag v*: construye y publica a PyPI/TestPyPI (instalable con pip).
• publish-packages.yml en tags v*: adjunta los paquetes al GitHub Release.
• main-only-from-develop.yml (workflow main-pr-source-check): valida metadata minima del PR hacia main.

Reglas de versionado automatico

• major: si algun commit trae BREAKING CHANGE, type(scope)!: o ramas/commits marcados como breaking/major.
• minor: si hay suficiente peso de feat/feature (proporcional frente a cambios patch) y no hay major.
• patch: cualquier otro caso.

Recomendado usar Conventional Commits para que el calculo de version sea preciso.

Secrets requeridos

• RELEASE_BOT_TOKEN (recomendado): PAT/fine-grained token para crear rama release/*, crear tag y publicar GitHub Release.
• RULESET_ADMIN_TOKEN (fallback): usado automaticamente si no existe RELEASE_BOT_TOKEN.
• PYPI_API_TOKEN: token de publicacion para PyPI.
• TEST_PYPI_API_TOKEN (opcional): token de publicacion para TestPyPI.