2024-11-28
9 min read

Lightweight Alternatives to Python Twisted for Async Networking

Lightweight Alternatives to Python Twisted for Async Networking

TLDR: For new projects, use asyncio (built into Python) with aiohttp for HTTP or Trio for complex concurrent applications. Twisted is feature-rich but heavyweight. Modern alternatives like asyncio, Trio, and Curio offer simpler APIs, better debugging, and lighter footprints while handling most async networking needs effectively.

Twisted has been the go-to framework for asynchronous network programming in Python for over two decades. It's feature-complete and battle-tested, but it's also complex, has a steep learning curve, and can feel heavy for simple projects. If you're looking for alternatives that are lighter weight or easier to work with, here are your options.

Built-in asyncio

Since Python 3.4, asyncio has been part of the standard library. It provides the core async/await syntax and event loop functionality without external dependencies.

Basic TCP Server with asyncio

Here's a simple echo server using asyncio:

import asyncio

async def handle_client(reader, writer):
    """Handle a single client connection."""
    addr = writer.get_extra_info('peername')
    print(f"Connection from {addr}")

    try:
        while True:
            # Read data from the client
            data = await reader.read(1024)
            if not data:
                break

            message = data.decode()
            print(f"Received: {message.strip()} from {addr}")

            # Echo it back
            writer.write(data)
            await writer.drain()

    except Exception as e:
        print(f"Error with {addr}: {e}")
    finally:
        writer.close()
        await writer.wait_closed()
        print(f"Disconnected: {addr}")

async def main():
    server = await asyncio.start_server(
        handle_client, 'localhost', 8888
    )

    addr = server.sockets[0].getsockname()
    print(f"Server listening on {addr}")

    async with server:
        await server.serve_forever()

# Run the server
asyncio.run(main())

This is significantly simpler than the equivalent Twisted code. The async/await syntax is intuitive if you're familiar with modern Python.

Why Choose asyncio

Pros:

  • Built into Python 3.4+ - no installation needed
  • Standard async/await syntax matches other languages
  • Growing ecosystem of async libraries
  • Good documentation and community support
  • Lower learning curve than Twisted

Cons:

  • Lower-level than Twisted - you build more yourself
  • Less mature protocol implementations
  • Error handling can be tricky
  • Debugging async code takes practice

Best for: General-purpose async programming where you want standard library support and modern Python syntax.

aiohttp: HTTP Made Easy

If you're building HTTP clients or servers, aiohttp is the most popular asyncio-based option:

from aiohttp import web
import asyncio

async def hello(request):
    """Simple HTTP handler."""
    name = request.match_info.get('name', 'World')
    return web.Response(text=f"Hello, {name}!")

async def websocket_handler(request):
    """WebSocket echo handler."""
    ws = web.WebSocketResponse()
    await ws.prepare(request)

    async for msg in ws:
        if msg.type == web.WSMsgType.TEXT:
            await ws.send_str(f"Echo: {msg.data}")
        elif msg.type == web.WSMsgType.ERROR:
            print(f'WebSocket error: {ws.exception()}')

    return ws

# Create application
app = web.Application()
app.router.add_get('/', hello)
app.router.add_get('/hello/{name}', hello)
app.router.add_get('/ws', websocket_handler)

# Run server
if __name__ == '__main__':
    web.run_app(app, host='localhost', port=8080)

For HTTP clients:

import aiohttp
import asyncio

async def fetch_multiple():
    """Fetch multiple URLs concurrently."""
    urls = [
        'https://api.github.com/users/github',
        'https://api.github.com/users/python',
        'https://api.github.com/users/microsoft',
    ]

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

async def fetch_url(session, url):
    """Fetch a single URL."""
    async with session.get(url) as response:
        data = await response.json()
        return data['name']

# Run
results = asyncio.run(fetch_multiple())
print(results)

Why choose aiohttp:

  • Simple, modern HTTP client and server
  • WebSocket support built-in
  • Middleware support for request/response processing
  • Good performance
  • Active development

Best for: HTTP APIs, web scraping, microservices, any HTTP-based networking.

Trio: Structured Concurrency

Trio takes a different approach to async programming with its "structured concurrency" model:

import trio

async def handle_client(stream):
    """Handle a single client connection."""
    print(f"Connection from {stream.socket.getpeername()}")

    async with stream:
        async for data in stream:
            print(f"Received: {data.decode().strip()}")

            # Echo back
            await stream.send_all(data)

async def main():
    """Run the echo server."""
    await trio.serve_tcp(handle_client, port=8888)

# Run the server
trio.run(main)

Trio's key innovation is its approach to cancellation and timeouts:

import trio
import httpx

async def fetch_with_timeout(url, timeout_seconds):
    """Fetch URL with automatic timeout and cleanup."""

    async with trio.open_nursery() as nursery:
        # Set a timeout for the entire operation
        with trio.move_on_after(timeout_seconds) as cancel_scope:
            async with httpx.AsyncClient() as client:
                response = await client.get(url)
                return response.text

        if cancel_scope.cancelled_caught:
            print(f"Request timed out after {timeout_seconds}s")
            return None

# Usage
result = trio.run(fetch_with_timeout, 'https://example.com', 5.0)

The "nursery" pattern makes it much harder to accidentally leak tasks or leave resources unclosed.

Why choose Trio:

  • Excellent error handling and cancellation
  • Harder to write buggy concurrent code
  • Great debugging tools
  • Clean, readable API
  • Strong focus on correctness

Cons:

  • Smaller ecosystem than asyncio
  • Different enough from asyncio that libraries aren't compatible
  • Less mature

Best for: Complex concurrent applications where correctness is critical, or when you're learning async programming and want guard rails.

Curio: Lightweight and Fast

Curio is similar to Trio but focuses on simplicity and performance:

from curio import run, spawn, tcp_server

async def echo_client(client, addr):
    """Handle a client connection."""
    print(f"Connection from {addr}")

    async with client:
        async for line in client.makefile('rb'):
            await client.sendall(line)

    print(f"Disconnected: {addr}")

async def main():
    """Start the echo server."""
    async with tcp_server('', 8888, echo_client) as server:
        print(f"Server listening on {server.address}")
        await server.wait_closed()

# Run
if __name__ == '__main__':
    run(main)

Curio is smaller and faster than asyncio in many benchmarks:

from curio import run, TaskGroup
import time

async def worker(n):
    """Simulate some work."""
    await curio.sleep(1)
    return n * 2

async def main():
    """Run multiple workers concurrently."""
    async with TaskGroup() as g:
        for i in range(10):
            await g.spawn(worker, i)

    # All tasks complete here
    print("All workers finished")

run(main)

Why choose Curio:

  • Lighter weight than asyncio
  • Clean, minimal API
  • Good performance
  • Excellent for teaching async concepts

Cons:

  • Smallest ecosystem
  • Less active development than Trio or asyncio
  • Fewer built-in protocols

Best for: Performance-critical applications, embedded systems, learning async programming.

gevent: Greenlet-Based Concurrency

gevent isn't truly async - it uses greenlets (lightweight threads) and monkey-patching to make synchronous code run concurrently:

from gevent import monkey
monkey.patch_all()  # Monkey-patch standard library

from gevent.server import StreamServer
import socket

def handle_client(sock, address):
    """Handle a client connection (looks synchronous!)."""
    print(f"Connection from {address}")

    fileobj = sock.makefile(mode='rb')
    for line in fileobj:
        sock.sendall(line)

    sock.close()
    print(f"Disconnected: {address}")

# Create server
server = StreamServer(('localhost', 8888), handle_client)
print("Server listening on localhost:8888")
server.serve_forever()

Notice there are no async or await keywords. gevent makes synchronous code concurrent through greenlets:

import gevent
from gevent import socket

def fetch_url(url):
    """Fetch a URL (looks synchronous but doesn't block other greenlets)."""
    sock = socket.create_connection(('example.com', 80))
    request = f"GET / HTTP/1.0\r\nHost: example.com\r\n\r\n"
    sock.sendall(request.encode())

    response = b''
    while True:
        data = sock.recv(1024)
        if not data:
            break
        response += data

    sock.close()
    return len(response)

# Run multiple fetches concurrently
jobs = [gevent.spawn(fetch_url, 'http://example.com') for _ in range(10)]
gevent.joinall(jobs)

results = [job.value for job in jobs]
print(f"Received {sum(results)} bytes total")

Why choose gevent:

  • Use synchronous code style with concurrent execution
  • Easy to add concurrency to existing sync code
  • Large ecosystem of compatible libraries
  • Proven at scale (used by many companies)

Cons:

  • Monkey-patching can cause subtle bugs
  • Harder to reason about flow control
  • Not compatible with truly async libraries
  • Debugging can be challenging

Best for: Adding concurrency to existing synchronous code, or when you prefer synchronous programming style but need concurrency.

Comparison Table

Here's a quick reference for choosing:

Framework      | Size      | Learning Curve | Ecosystem | Speed
---------------|-----------|----------------|-----------|-------
asyncio        | Medium    | Medium         | Large     | Good
aiohttp        | Small     | Easy           | Medium    | Good
Trio           | Small     | Easy-Medium    | Small     | Good
Curio          | Tiny      | Easy           | Tiny      | Excellent
gevent         | Small     | Easy           | Large     | Good
Twisted        | Large     | Hard           | Huge      | Excellent

Dependencies:
asyncio        | None (stdlib)
aiohttp        | pip install aiohttp
Trio           | pip install trio
Curio          | pip install curio
gevent         | pip install gevent
Twisted        | pip install twisted

Real-World Example: WebSocket Chat Server

Here's the same chat server in different frameworks to show the differences:

asyncio + websockets

import asyncio
import websockets

connected_clients = set()

async def chat_handler(websocket, path):
    """Handle a WebSocket connection."""
    connected_clients.add(websocket)
    try:
        async for message in websocket:
            # Broadcast to all clients
            await asyncio.gather(
                *[client.send(message) for client in connected_clients]
            )
    finally:
        connected_clients.remove(websocket)

async def main():
    async with websockets.serve(chat_handler, "localhost", 8765):
        await asyncio.Future()  # Run forever

asyncio.run(main())

Trio + trio-websocket

import trio
from trio_websocket import serve_websocket

connected_clients = set()

async def chat_handler(request):
    """Handle a WebSocket connection."""
    ws = await request.accept()
    connected_clients.add(ws)

    try:
        async for message in ws:
            # Broadcast to all clients
            for client in connected_clients:
                await client.send_message(message)
    finally:
        connected_clients.remove(ws)

async def main():
    await serve_websocket(chat_handler, "localhost", 8765, ssl_context=None)

trio.run(main)

The code structure is nearly identical, showing how similar these frameworks are in practice.

Migration Path from Twisted

If you have existing Twisted code and want to migrate:

  1. For new features: Write them in asyncio or Trio instead of adding to Twisted code
  2. For HTTP services: Replace with aiohttp (easiest migration)
  3. For complex state machines: Consider Trio for better structure
  4. For performance-critical paths: Test Curio as a replacement

You can run Twisted and asyncio in the same process during migration:

from twisted.internet import asyncioreactor
asyncioreactor.install()

from twisted.internet import reactor
import asyncio

# Now you can use both Twisted and asyncio in the same application
async def asyncio_code():
    await asyncio.sleep(1)
    print("asyncio code running")

# Run asyncio code from Twisted
asyncio.ensure_future(asyncio_code())

# Start Twisted reactor
reactor.run()

For most new Python projects, start with asyncio for general networking or aiohttp for HTTP. Consider Trio if you need better structured concurrency or are building something complex. Twisted is still excellent for specific use cases (IRC bots, mail servers, legacy protocol implementations), but modern alternatives offer simpler APIs and better developer experience for common networking tasks.

Published: 2024-11-28|Last updated: 2024-11-28T13:00:00Z

Found an issue?