aiosocket-python 0.1.1

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 — inspired by aiogram and FastAPI, 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()

asyncio.run(main())

Filters

Filter routes connections by matching raw data against a regex, JSON fields, or both. Multiple filters on a single handler are evaluated in order and short-circuit on the first mismatch.

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 a JSON field value
login_filter = Filter(action="login")

# Combine: must match both the pattern and the JSON field
strict_filter = Filter(pattern=r"^\{.*\}$", 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

Every incoming connection runs all middleware whose filters match, in registration order, before any callback is invoked. Middleware can populate a shared data dict that is forwarded as keyword arguments 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())

Only the first callback whose filters match is invoked; subsequent callbacks are skipped.

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) — register a connection handler; first match wins
• .middleware(*filters) — register middleware; all matches run before the callback
• .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 and concurrency-safe
• .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