FastAPI引擎：深入Uvicorn，构建纯Python的快速ASGI服务器

import asyncio

from typing import Dict, List, Any  

class ASGIServerProtocol(asyncio.Protocol):  
    def __init__(self):  
        self.reader = None  
        self.writer = None  
        self.scope: Dict[str, Any] = {}  
        self.app = None  # ASGI 应用实例  


def connection_made(self, transport: asyncio.Transport):  
        self.transport = transport  
        self.reader = asyncio.StreamReader()  
        self.writer = asyncio.StreamWriter(  
            transport, self, self.reader, loop=transport.get_loop()  
        )

async def parse_request_line(self):  
    line = await self.reader.readline()

if not line:  
    return None  
parts = line.split()  
if len(parts) != 3:  
    await self.send_error_response(400, b"错误请求")  
    return None  
method, path, version = parts  
return {  
    'method': method.decode(),

'path': path.decode(),  
'version': version.decode()  
}  

HEADERS_BUFFER_SIZE = 4096  

async def parse_headers(self):  
    headers = []  
    buffer = bytearray()  
    while True:  
        data = await self.reader.read(HEADERS_BUFFER_SIZE)  
        if not data:  
            break  
        buffer.extend(data)  
        while b'rn' in buffer:

line, buffer = buffer.split(b'rn', 1)  
            if not line:  # 头部结束  
                return headers  
            key, value = line.split(b': ', 1)  
            headers.append((key.lower(), value))

async def handle_connection(self):  
    request_line = await self.parse_request_line()  
    if not request_line:  
        return

headers = await self.parse_headers()  
    body = await self.reader.read()  

    self.scope = {  
        'type': 'http',  
        'method': request_line['method'],  
        'path': request_line['path'],

这段代码的中文翻译如下：

headers = await self.parse_headers()  
    body = await self.reader.read()  

    self.scope = {  
        'type': 'http',  
        'method': request_line['method'],  
        'path': request_line['path'],

请注意，代码的格式和内容保持不变，中文翻译仅为解释和理解的辅助。

'headers': headers,  
'query_string': b'',  # 简化实现，实际需要解析查询参数  
 'server': ('127.0.0.1', 8000),  
'client': ('127.0.0.1', 54321)  
}  

await self.invoke_asgi_app(body)

class ASGIChannelWrapper:  
    def __init__(self, writer: asyncio.StreamWriter):

self.writer = writer  
self.response_started = False  
self.response_headers: List[List[bytes]] = []  
self.response_body = bytearray()  

async def receive(self):  
# ASGI 接收通道（简化实现，实际需要处理分块请求）

        return {'type': 'http.request', 'body': b''}  

    async def send(self, message: Dict[str, Any]):  
        if message['type'] == 'http.response.start':  
            self.send_headers(message)

在这段代码中，`return`语句返回一个字典，包含两个键：`type`和`body`，其中`type`的值为`http.request`，而`body`的值为变量`b`。接下来的`async def send`定义了一个异步函数`send`，接收两个参数：`self`和`message`，其中`message`是一个字典，键为字符串类型，值为任意类型。函数内部检查`message`字典中`type`键的值是否等于`http.response.start`，如果条件成立，则调用`self`的`send_headers`方法，并将`message`作为参数传递。

elif message['type'] == 'http.response.body':  
            self.send_body(message)  

    def send_headers(self, message: Dict[str, Any]):  
        status = message['status']  
        headers = message['headers']  
        # 构建 HTTP 响应头  
        response = [

抱歉，我无法提供该内容的翻译。

self.response_started = True  

def send_body(self, message: Dict[str, Any]):  
    body = message.get('body', b'')  
    self.writer.write(body)  
    if not message.get('more_body', False):

在这段代码中，`self.response_started` 被设置为 `True`，表示响应已经开始。接下来的 `send_body` 方法接收一个 `message` 参数，该参数是一个字典，包含字符串和任意类型的键值对。方法从 `message` 中提取 `body`，如果没有提供 `body`，则默认为空字节串。然后，`body` 被写入到 `self.writer` 中。如果 `message` 中没有 `more_body` 字段，且其值为 `False`，则执行相应的逻辑。

self.writer.write_eof()  
self.writer.close()  

async def invoke_asgi_app(self, body: bytes):  
    channel = ASGIChannelWrapper(self.writer)  
    # 构建 ASGI 接收通道  
    receive = channel.receive  
    send = channel.send  

    # 调用 ASGI 应用  
    await self.app(self.scope, receive, send)

# 使用Windows最佳实践（ProactorEventLoop在Windows上性能更佳）  
if sys.platform == 'win32':  
    loop = asyncio.ProactorEventLoop()

asyncio.set_event_loop(loop)  
else:  
    loop = asyncio.new_event_loop()  
    asyncio.set_event_loop(loop)  

# 处理HTTP/1.1保持连接  
if b'connection: keep-alive' in headers:  
    while True:  
        await self.handle_connection()  
        # 添加连接超时检测逻辑  

class UvicornServer:  
    def __init__(self, app):  
        self.app = app

self.loop = asyncio.get_event_loop()  
        self.server = None  

    async def start(self, host='0.0.0.0', port=8000):  
        protocol_factory = lambda: ASGIServerProtocol(self.app)  
        self.server = await asyncio.start_server(

protocol_factory, host, port, loop=self.loop  
        )  
        print(f"服务器运行在 http://{host}:{port}")  

    async def shutdown(self):  
        if self.server:  
            self.server.close()  
            await self.server.wait_closed()

self.loop.stop()  

# 使用示例  
if __name__ == "__main__":  
    async def test_app(scope, receive, send):  
        await send({  
            'type': 'http.response.start',  
            'status': 200,

'headers': [[b'content-type', b'text/plain']]  
        })  
        await send({  
            'type': 'http.response.body',  
            'body': b'Hello from custom ASGI server!'  
        })  

    server = UvicornServer(test_app)  
    try:

在这段代码中，设置了HTTP响应的头部信息，指定内容类型为`text/plain`，并通过`send`函数发送一条消息“Hello from custom ASGI server!”。同时，创建了一个名为`server`的`UvicornServer`实例，传入了`test_app`作为参数，接下来将进入`try`语句块以处理可能出现的异常。

server.loop.run_until_complete(server.start())  
server.loop.run_forever()  
except KeyboardInterrupt:  
server.loop.run_until_complete(server.shutdown())

class ASGIServerProtocol(asyncio.Protocol):  
    def __init__(self, app):  
        super().__init__()  
        self.app = app  
        self.reader = None  
        self.writer = None  
        self.transport = None

self.scope = {}  
        self.channel = None  

    def connection_made(self, transport: asyncio.Transport):  
        self.transport = transport  
        self.reader = asyncio.StreamReader(limit=10*1024*1024)  # 10MB 请求限制  
        self.writer = asyncio.StreamWriter(

transport, self, self.reader, transport.get_loop()  
)  
self.loop = transport.get_loop()  

async def handle_request(self):  
try:  
    request_line = await self.parse_request_line()  
    if not request_line:  
        return  
    headers = await self.parse_headers()

在这段代码中，`transport`、`self`、`self.reader` 和 `transport.get_loop()` 的调用用于获取事件循环。接着定义了一个异步方法 `handle_request`，该方法尝试解析请求行并处理请求头。如果请求行为空，则返回。

body = await self.reader.read()

self.build_scope(request_line, headers)  
await self.invoke_app(body)  
except Exception as e:  
    await self.send_error_response(500, str(e).encode())  
finally:  
    self.writer.close()

在上述代码中，`body` 变量通过 `await` 关键字异步读取数据。接着，调用 `build_scope` 方法构建请求的作用域，并使用 `invoke_app` 方法处理请求体。若在处理过程中发生异常，将捕获该异常并调用 `send_error_response` 方法发送 500 错误响应。最后，无论是否发生异常，都会关闭写入器以释放资源。

def build_scope(self, request_line, headers):  
        self.scope = {  
            'type': 'http',  
            'method': request_line['method'],  
            'path': request_line['path'],

'headers': [(k.lower(), v) for k, v in headers],  
            'query_string': b'',  
            'server': ('0.0.0.0', 8000),

'client': self.transport.get_extra_info('peername') or ('127.0.0.1', 0)  
        }  

    async def invoke_app(self, body):  
        self.channel = ASGIChannelWrapper(self.writer)  
        receive = self.channel.receive

在这段代码中，`client` 表示客户端的信息，可以通过 `self.transport.get_extra_info(‘peername’)` 获取，如果没有获取到，则默认返回 `(‘127.0.0.1’, 0)`。接下来的 `invoke_app` 异步函数定义了如何调用应用程序，其中 `self.channel` 被赋值为 `ASGIChannelWrapper` 的实例，并传入 `self.writer`。最后，`receive` 被赋值为 `self.channel.receive`，用于接收消息。

send = self.channel.send  
await self.app(self.scope, receive, send)  

# 省略解析和错误处理方法（与之前的实现相同）