aiosocket-python 0.1.0

A lightweight async TCP server framework built on asyncio

aiosocket

A lightweight async TCP server framework built on Python's asyncio. Provides a decorator-based API for defining handlers, middleware, and filter-based routing — similar in spirit to Flask, but for raw TCP connections.

Requirements

Python 3.10+

Installation

pip install aiosocket-python

Quick Start

import asyncio
import aiosocket

server = aiosocket.AsyncTCPServer(host="127.0.0.1", port=8888)

@server.callback()
async def echo(reader: aiosocket.Reader, writer: aiosocket.Writer):
    data = await reader.data()
    await writer.send(data)

async def main():
    await server.start()
    await asyncio.Event().wait()  # run forever

asyncio.run(main())

Filters

Use Filter to route connections based on raw data patterns or JSON fields.

from aiosocket import AsyncTCPServer, Filter

server = AsyncTCPServer("127.0.0.1", 8888)

# Match raw bytes with a regex pattern
text_filter = Filter(pattern=r"^HELLO.*")

# Match JSON fields
login_filter = Filter(action="login")

@server.callback(login_filter)
async def handle_login(reader, writer):
    payload = await reader.json()
    await writer.send({"status": "ok", "user": payload["user"]})

@server.callback(text_filter)
async def handle_hello(reader, writer):
    await writer.send(b"Hello to you too!\n")

Middleware

Middleware runs before callbacks and can populate a shared data dict passed as **kwargs to the matched callback.

@server.middleware()
async def log_request(reader, writer, data):
    raw = await reader.data()
    print(f"Received {len(raw)} bytes")

@server.middleware(login_filter)
async def authenticate(reader, writer, data):
    payload = await reader.json()
    data["user"] = payload.get("user")

@server.callback(login_filter)
async def greet(reader, writer, user=None):
    await writer.send(f"Welcome, {user}!".encode())

Routers

Split your handlers across modules using Router, then mount them on the server.

from aiosocket import Router

router = Router()

@router.callback()
async def handle(reader, writer):
    await writer.send(b"OK\n")

# in your main file:
server.include_router(router)

API Reference

AsyncTCPServer(host, port)

• .callback(*filters) — decorator that registers a connection handler
• .middleware(*filters) — decorator that registers middleware
• .include_router(router) — merge a Router's handlers into this server
• .start() — start listening (coroutine)
• .stop() — gracefully shut down (coroutine)

Reader

• .data() -> bytes — read up to 64 KiB; result is cached
• .json() -> dict — parse data as JSON; result is cached

Writer

• .send(data: bytes | str | dict) — write and drain; dicts are JSON-serialized
• .close() — close the connection

Filter(pattern=None, **json_fields)

• pattern — regex matched against the raw decoded bytes
• **json_fields — key/value pairs matched against the parsed JSON body

Router

• .callback(*filters) — same as AsyncTCPServer.callback
• .middleware(*filters) — same as AsyncTCPServer.middleware

License

MIT