py-netty :rocket:

py-netty is a lightweight Netty-style, event-driven TCP networking framework for Python, focused on non-blocking socket programming with callback-based channel handlers.

Features

• Netty-style programming model: build TCP clients and servers with Bootstrap, ServerBootstrap, EventLoopGroup, Channel, and callback-based ChannelHandler APIs.
• Non-blocking TCP I/O: sockets are managed by selector-backed event loops, keeping application code event-driven instead of thread-per-connection.
• Thread-confined socket operations: recv and write are performed from the owning I/O thread, making channel behavior predictable under concurrency.
• Backpressure-aware writes: low and high water marks expose channel writability changes, with defaults of 32 KiB and 64 KiB.
• Adaptive read buffering: read buffer sizes grow or shrink based on recent traffic patterns to balance throughput and memory usage.
• TLS support for clients and servers: wrap TCP channels with SSL/TLS, including optional certificate verification and custom SSL context hooks.
• Cross-platform selector support: uses the best selector available on each platform, such as epoll on Linux, kqueue on macOS, and select on Windows.

Installation

pip install py-netty

Getting Started

Start an echo server:

from py_netty import ServerBootstrap
ServerBootstrap().bind(address='0.0.0.0', port=8080).close_future().sync()

Start an echo server (TLS):

from py_netty import ServerBootstrap
ServerBootstrap(certfile='/path/to/cert/file', keyfile='/path/to/key/file').bind(address='0.0.0.0', port=9443).close_future().sync()

As TCP client:

from py_netty import Bootstrap, ChannelHandlerAdapter


class HttpHandler(ChannelHandlerAdapter):
    def channel_read(self, ctx, buffer):
        print(buffer.decode('utf-8'))
        

remote_address, remote_port = 'www.google.com', 80
b = Bootstrap(handler_initializer=HttpHandler)
channel = b.connect(remote_address, remote_port).sync().channel()
request = f'GET / HTTP/1.1\r\nHost: {remote_address}\r\n\r\n'
channel.write(request.encode('utf-8'))
input() # pause
channel.close()

As TCP client (TLS):

from py_netty import Bootstrap, ChannelHandlerAdapter


class HttpHandler(ChannelHandlerAdapter):
    def channel_read(self, ctx, buffer):
        print(buffer.decode('utf-8'))
        

remote_address, remote_port = 'www.google.com', 443
b = Bootstrap(handler_initializer=HttpHandler, tls=True, verify=True)
channel = b.connect(remote_address, remote_port).sync().channel()
request = f'GET / HTTP/1.1\r\nHost: {remote_address}\r\n\r\n'
channel.write(request.encode('utf-8'))
input() # pause
channel.close()

TCP port forwarding:

from py_netty import ServerBootstrap, Bootstrap, ChannelHandlerAdapter, EventLoopGroup


class ProxyChannelHandler(ChannelHandlerAdapter):

    def __init__(self, remote_host, remote_port, client_eventloop_group):
        self._remote_host = remote_host
        self._remote_port = remote_port
        self._client_eventloop_group = client_eventloop_group
        self._client = None

    def _client_channel(self, ctx0):

        class __ChannelHandler(ChannelHandlerAdapter):
            def channel_read(self, ctx, bytebuf):
                ctx0.write(bytebuf)

            def channel_inactive(self, ctx):
                ctx0.close()

        if self._client is None:
            self._client = Bootstrap(
                eventloop_group=self._client_eventloop_group,
                handler_initializer=__ChannelHandler
            ).connect(self._remote_host, self._remote_port).sync().channel()
        return self._client

    def exception_caught(self, ctx, exception):
        ctx.close()

    def channel_read(self, ctx, bytebuf):
        self._client_channel(ctx).write(bytebuf)

    def channel_inactive(self, ctx):
        if self._client:
            self._client.close()


proxied_server, proxied_port = 'www.google.com', 443
client_eventloop_group = EventLoopGroup(1, 'ClientEventloopGroup')
sb = ServerBootstrap(
    parent_group=EventLoopGroup(1, 'Acceptor'),
    child_group=EventLoopGroup(1, 'Worker'),
    child_handler_initializer=lambda: ProxyChannelHandler(proxied_server, proxied_port, client_eventloop_group)
)
sb.bind(port=8443).close_future().sync()

Event-driven callbacks

Create handler with callbacks for interested events:

from py_netty import ChannelHandlerAdapter


class MyChannelHandler(ChannelHandlerAdapter):
    def channel_active(self, ctx: 'ChannelHandlerContext') -> None:
        # invoked when channel is active (TCP connection ready)
        pass

    def channel_read(self, ctx: 'ChannelHandlerContext', msg: Union[bytes, socket.socket]) -> None:
        # invoked when there is data ready to process
        pass

    def channel_inactive(self, ctx: 'ChannelHandlerContext') -> None:
        # invoked when channel is inactive (TCP connection is broken)
        pass

    def channel_registered(self, ctx: 'ChannelHandlerContext') -> None:
        # invoked when the channel is registered with a eventloop
        pass

    def channel_unregistered(self, ctx: 'ChannelHandlerContext') -> None:
        # invoked when the channel is unregistered from a eventloop
        pass

    def channel_handshake_complete(self, ctx: 'ChannelHandlerContext') -> None:
        # invoked when ssl handshake is complete, this only applies to client side
        pass

    def channel_writability_changed(self, ctx: 'ChannelHandlerContext') -> None:
        # invoked when pending data > high water mark or < low water mark
        pass

    def exception_caught(self, ctx: 'ChannelHandlerContext', exception: Exception) -> None:
        # invoked when there is any exception raised during process
        pass

Benchmark

The current benchmark uses the local echo performance runner in integration_tests/perf_echo.py. By default, each case starts an in-process localhost echo server for the selected engine. You can also provide --port with optional --host to run the same framed clients against a separately deployed echo server. The runner validates every echo and reports throughput, message rate, latency, and connection ramp-up time.

The following results were collected locally with:

python integration_tests/perf_echo.py --case all --engine all --timeout 20 --json
python integration_tests/perf_echo.py --case high_connection_scaling --engine all --timeout 30 --json

Environment: macOS 26.5 arm64, Python 3.12.10.

The default suite covers latency, payload throughput, backpressure, and moderate concurrency. It is useful for comparing broad behavior, not for declaring one engine universally faster.

High Connection Scaling

The high connection-count suite stresses 128, 256, and 512 concurrent localhost connections with 20 messages per connection and 1 KiB payloads. It highlights where py-netty's event-loop model pulls ahead of the one-thread-per-connection threaded implementation.

In this run, py-netty kept a much steadier message rate as connection count increased, while the threaded implementation degraded more quickly. Compared with threaded sockets, py-netty delivered 24% higher message rate at 128 connections, 48% higher at 256 connections, and 57% higher at 512 connections. This makes the 256-connection case the clearest inflection point for the threaded approach in this local test. asyncio is included as a standard library event-loop baseline and remained the fastest engine by raw message rate in these high-connection cases.

Metrics are informational and environment-dependent. The comparison uses three local implementations: py-netty, Python asyncio, and blocking sockets with one thread per connection (threaded). The performance runner fails only on functional problems such as missing echoes, payload mismatches, connection failures, or timeouts.

Throughput

Message Rate

Latency

Connection Ramp-up