pyasynchat
pyasynchat is a re-implementation of Python's `asynchat` standard library module, which was removed in Python 3.12. It provides a compatible API for asynchronous event-driven network communication, built upon `pyasyncore`. The current version is 1.0.5, and releases are made as needed to ensure compatibility with newer Python versions or address bugs.
Warnings
- breaking The original `asynchat` standard library module was removed from Python 3.12 and later versions.
- breaking Applications using `asynchat` must update their import statements to use `pyasynchat`.
- gotcha `pyasynchat` relies on `pyasyncore` for the underlying asynchronous event loop (`asyncore.loop`). Both packages must be installed for `pyasynchat` to function correctly.
- gotcha As a re-implementation of a removed standard library module, while aiming for high compatibility, subtle behavioral differences might exist compared to the original CPython standard library versions of `asynchat` and `asyncore`.
Install
-
pip install pyasynchat -
pip install pyasynchat pyasyncore
Imports
- async_chat
from pyasynchat import async_chat
- simple_producer
from pyasynchat import simple_producer
Quickstart
import socket
import pyasyncore
from pyasynchat import async_chat, simple_producer # simple_producer is imported but not used for brevity in this example
class EchoHandler(async_chat):
def __init__(self, sock):
super().__init__(sock)
self.set_terminator(b'\n')
self.data = []
def collect_incoming_data(self, data):
self.data.append(data)
def found_terminator(self):
message = b"".join(self.data).strip()
print(f"Received: {message.decode()}")
self.push(message + b'\n') # Echo back with a newline
self.data = []
class EchoServer(pyasyncore.dispatcher):
def __init__(self, host, port):
super().__init__()
self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
self.set_reuse_addr()
self.bind((host, port))
self.listen(5)
print(f"Listening on {host}:{port}")
def handle_accept(self):
pair = self.accept()
if pair is not None:
sock, addr = pair
print(f"Incoming connection from {addr}")
EchoHandler(sock)
if __name__ == "__main__":
# The loop function must be imported from pyasyncore, not pyasynchat.
# Ensure pyasyncore is also installed: pip install pyasyncore
server = EchoServer('localhost', 8080)
try:
pyasyncore.loop()
except KeyboardInterrupt:
print("Server stopped.")
# To test: In another terminal, run: nc localhost 8080