Lynkio · Real‑Time Event Engine

Lynkio is a lightweight, high‑performance framework for building real‑time web applications using only the Python standard library. It combines a native HTTP router, a full WebSocket implementation (RFC6455), a UDP datagram server, and an innovative AUTO mode that serves TCP (HTTP/WebSocket) and UDP concurrently on the same port. Designed for simplicity and control, Lynkio has no external dependencies except optional integration with soketDB for logging.

The Lynk class is the main entry point. It configures the server and exposes all routing decorators and methods.

from lynkio import Lynk
app = Lynk(host="0.0.0.0", port=8080, protocol="AUTO", rate_limit=30, debug=True)

The server is started with app.run() which installs signal handlers for graceful shutdown.

@app.get(path), @app.post(path), @app.put(path), @app.delete(path), @app.patch(path), @app.route(path, methods).
Path parameters: /user/<user_id> or /post/{post_id}. They become keyword arguments in the handler.

@app.get("/users/<user_id>/posts/<post_id>")
async def get_post(req, user_id, post_id):
    return {"user": user_id, "post": post_id}

req.method, req.path, req.headers (dict), req.body (bytes), req.client_ip, req.query_params (dict), req.cookies (dict), await req.json(), await req.form().

• json_response(data, status=200) – returns JSON response.
• redirect(location, status=302) – HTTP redirect.
• abort(status_code, message) – raises HTTPError.
• send_file(filepath, as_attachment=False, cache_control=None, ...) – streams a file with support for Content-Disposition, Last‑Modified, Cache‑Control.
• render_template(template_name, context=None, template_dir="templates") – basic template with {{ variable.nested }} syntax.
• FileResponse and StreamingResponse classes for custom streaming.

from lynkio import send_file, render_template

@app.get("/download")
async def download(req):
    return send_file("report.pdf", as_attachment=True, cache_control="max-age=3600")

@app.get("/page")
async def page(req):
    return render_template("index.html", {"title": "Lynkio"})

app.static(prefix, directory) – serves files from a directory, prevents directory traversal.
app.group(prefix) – returns a RouteGroup that can have its own middleware and routes.

app.static("/assets", "public")
api = app.group("/api/v1")
@api.get("/health")
async def health(req):
    return {"status": "ok"}

Add functions to app._http_middleware (list) – each receives req and can return None or a response (bytes).
Built‑in CORS: app.enable_cors(allowed_origins, allow_credentials) adds the middleware and handles preflight.

app.enable_cors(["https://example.com"], allow_credentials=True)

WebSocket upgrade is automatic when the client sends an Upgrade: websocket header. Lynkio implements RFC6455 with fragmentation, ping/pong heartbeat, and binary frames.

Inside event handlers, client is a Connection instance with:
client.id (unique UUID), client.session (dict for custom data), client.rooms (set of rooms), await client.send(payload, text=True), await client.ping(), await client.close(code, reason).

@app.on(event_name) – receives JSON messages with {"event": ..., "data": ...}.
@app.on_binary – receives raw binary payloads.
@app.on_binary_event(name) – receives named binary events (client sends a 2‑byte length header + name).

@app.on("chat")
async def chat(client, data):
    await app.emit_to_room(data["room"], "chat", data["text"])

@app.on_binary_event("image")
async def image_stream(client, payload: bytes):
    print(f"Received image chunk {len(payload)} bytes")

@app.middleware – each middleware receives (client, event, data) and can return modified data or raise StopProcessing to reject the event.

@app.middleware
async def auth_mw(client, event, data):
    if event == "admin" and not client.session.get("auth"):
        raise StopProcessing
    return data

Rooms allow grouping clients. Methods: app.join_room(client_id, room), app.leave_room(client_id, room), app.emit_to_room(room, event, data, exclude=None) (batched to avoid event loop overload), app.get_room_clients(room) -> Set[str]. Broadcasting to all clients: await app.emit(event, data) (client_id=None).

@app.on("join")
async def join_room(client, data):
    app.join_room(client.id, data["room"])
    await app.emit_to_room(data["room"], "system", f"{client.id} joined")

@app.udp(path) – registers a handler for UDP JSON messages. The message must contain a "path" field, optionally "data" and "client_id" for rate limiting. The handler receives a Request object (method="UDP"). Response is sent back as a UDP datagram. AUTO mode runs both TCP (HTTP/WebSocket) and UDP on the same port simultaneously.

@app.udp("/sensor")
async def sensor(req):
    payload = await req.json()
    return {"status": "ok", "echo": payload}

# Client example:
# echo '{"path":"/sensor","data":{"temp":22.5},"client_id":"sensor1"}' | nc -u localhost 8765

UDP supports the same HTTP middleware chain, rate limiting (based on client_id or IP:port), and max payload enforcement.

@app.task – registers an async function that starts when the server runs and should loop until app._running becomes False.
@app.schedule(interval) – runs the coroutine periodically every interval seconds (cron‑like). Both are automatically cancelled on graceful shutdown.

@app.task
async def monitor():
    while app._running:
        await asyncio.sleep(30)
        print(f"Active connections: {len(app._clients)}")

@app.schedule(3600)
async def hourly_cleanup():
    await app.query_database("logs", "DELETE FROM old_entries")

Lynkio includes soketDB, a lightweight file‑based database with async support. Set enable_database=True and optionally provide database_config (primary_storage, backups, cloud providers).

app = Lynk(enable_database=True)
db = app.create_database("myapp", create_log_table=True, auto_sync_log=True)

When create_log_table=True, three tables are created: http_logs, wss_logs, runtime_logs. If auto_sync_log=True, every HTTP request, WebSocket message, connect/disconnect, and runtime event is automatically inserted asynchronously.

Manual logging: await app.add_log(table, **kwargs) where table is 'http', 'wss', or 'runtime'.
Distributed queries: await app.query_database(db_name, sql, params=None) – runs on any registered soketDB instance in a thread executor (non‑blocking).

await app.add_log("runtime", level="INFO", message="Custom event", source="auth")
rows = await app.query_database("myapp", "SELECT * FROM http_logs WHERE status_code = $1", (200,))

Include <script src="/lynkio/client.js"></script> (set serve_client=True). The global LynkClient class provides auto‑reconnect, binary events, rooms, and task scheduling.

const client = new LynkClient("ws://localhost:8765", { maxReconnectAttempts: 10 });
client.on("message", (data) => console.log(data));
client.onBinary("audio", (payload) => { /* Uint8Array */ });
client.connect().then(() => {
  client.joinRoom("lobby");
  client.emit("chat", { text: "hi" });
  client.sendBinaryEvent("image", new Uint8Array([1,2,3]));
  let tid = client.createTask(() => client.emit("ping", {}), 5000);
});
client.close();

Full async client with WebSocket, HTTP, and UDP support. Import from lynkio.

from lynkio import LynkClient
import asyncio

async def main():
    client = LynkClient("localhost", 8765)
    await client.ws.connect()
    client.on("greet", print)
    await client.ws.emit("ping", "hello")
    await client.ws.send_binary_event("data", b'\x01\x02')
    # HTTP
    status, headers, body = await client.http.get("/api/status")
    # UDP
    resp = await client.udp.send(b'{"path":"/ping"}')
    await client.ws.close()
asyncio.run(main())

Separate classes: WebSocketClient, HTTPClient, UDPClient are also available.

Plugin system: app.use(plugin) where plugin is a callable that receives the app instance – can add routes, middleware, etc.

def metrics_plugin(app):
    @app.get("/metrics")
    async def metrics(req):
        return {"clients": len(app._clients)}
app.use(metrics_plugin)

Internal HTTP client: await app.fetch(url, method, headers, body, json, timeout) – returns a FetchResponse with .json(), .text(), .content.

resp = await app.fetch("https://api.example.com/data")
data = await resp.json()

Internal events: @app.on_internal(event) for "connect", "disconnect", "subscribe", "unsubscribe", "message" – useful for analytics.

@app.on_internal("connect")
async def on_connect(client, _):
    print(f"New client {client.id}")