Asynchronous socket server implementation from raw sockets

Question

I've rewritten my HTTP server several times over the past year, because I tend to encounter an issue midway through an implementation realizing I didn't abstract, decompose, or decouple certain methods properly. My primary goal with these rewrites is to slowly edge towards a modular and small, but still precise implementation of the HTTP protocol; with the vision of high vertical & horizontal scalability in mind.

So, to begin the implementation I create the abstract base class defining the generic protocol, with which the socket server is going to interface with when socket events occur:

class BaseProtocol(ABC):
    def __init__(self, server, sk_client: socket.socket, mqueue: queue.Queue):
        self.server = server
        self.trans = sk_client
        self._mqueue = mqueue

    @abstractmethod
    def on_new_connection(self, address: tuple[str, int]) -> None:
        pass

    @abstractmethod
    def on_data_received(self, data: bytes) -> None:
        pass

    @abstractmethod
    def on_connection_closed(self) -> None:
        pass

    def close_socket_connection(self) -> None:
        self.server._close_connection(self.trans)

    def send(self, data: Union[str, bytes]) -> None:
        if isinstance(data, str):
            data = data.encode()
        self._mqueue.put_nowait(data)

In terms of minutiae, I think the style of adding @staticmethod to abstract-methods may appear cleaner. Then the implementation of the AsyncSocketServer:

class AsyncSocketServer:
    RX_CHUNKSIZE = 1024

    def __init__(self, host: str, port: int, prot: BaseProtocol,
            *, backlog=15):
        self.host = host
        self.port = port
        self.prot = prot
        self.backlog = backlog

        self.socket = socket.socket()
        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
        self.socket.bind((host, port))
        self.socket.listen(backlog)
        self.socket.setblocking(False)

        self._socket_rwx = {
            "inputs": [self.socket],
            "outputs": []
        }
        self._socket_clients = {}
        self._socket_priority = []

    def _close_connection(self, sockfd: socket.socket,
            *, _bypass=False) -> None:
        prot = self._socket_clients[sockfd]
        if not _bypass and not prot._mqueue.empty():
            self._socket_priority.append({
                "socket": sockfd,
                "prot": prot,
                "callback": lambda: self._close_connection(
                    sockfd, _bypass=True
                )
            })
            return
        self._socket_rwx['inputs'].remove(sockfd)
        self._socket_rwx['outputs'].remove(sockfd)
        prot.on_connection_closed()
        del self._socket_clients[sockfd]
        sockfd.shutdown(socket.SHUT_RDWR)
        sockfd.close()

    def accept_clients(self) -> None:
        while self._socket_rwx['inputs']:
            while self._socket_priority:
                pr_sockinfo = self._socket_priority.pop(0)
                pr_sockinfo['socket'].send( # assume writeable
                    pr_sockinfo['prot']._mqueue.get_nowait()
                )
                pr_sockinfo['callback']()

            try:
                readable, writeable, exceptional = select.select(
                    self._socket_rwx['inputs'], self._socket_rwx['outputs'],
                    self._socket_rwx['inputs']
                )
            except KeyboardInterrupt:
                logging.info("caught keyboard interrupt, closing all clients")
                for sockfd in self._socket_clients:
                    self._close_connection(sockfd, _bypass=True)
                return

            for sk_exc in exceptional:
                self._close_connection(sk_exc)

            for sk_writeable in writeable:
                if (proto := self._socket_clients.get(sk_writeable)) is None:
                    continue
                if proto._mqueue.empty():
                    continue
                sk_writeable.send(proto._mqueue.get_nowait())

            for sk_readable in readable:
                if sk_readable is self.socket:
                    sk_client, cl_addr = self.socket.accept()
                    sk_client.setblocking(False)

                    self._socket_rwx['inputs'].append(sk_client)
                    self._socket_rwx['outputs'].append(sk_client)

                    self._socket_clients[sk_client] = \
                        (cl := self.prot(self, sk_client, queue.Queue()))
                    cl.on_new_connection(cl_addr)
                    continue

                sk_buffer = b""
                try:
                    while (fragment := sk_readable.recv(
                            AsyncSocketServer.RX_CHUNKSIZE)):
                        sk_buffer += fragment
                except BlockingIOError:
                    pass
                
                if not sk_buffer:
                    self._close_connection(sk_readable)
                    continue
                self._socket_clients[sk_readable].on_data_received(
                    sk_buffer
                )

My code adheres to a strict 80 character line-break, and in terms of documentation, I refrain from commenting most of any code since I believe in the most critical parts (in terms of maintanability), the code is clear to anyone reading, therefore comments would just insert noise.

So, I'm just wondering if there's any ideas that I could consider with AsyncSocketServer to make it more production-ready, whether in terms of easier readability in certain points, or regions of code that I can pull out to make it easier on the eyes, or anything.