使用 asyncio 启动一个简单的 Python 文件服务器

Question

使用 python3，我可以启动一个简单的 web 服务器，它为当前目录中的文件提供服务

from functools import partial
from http.server import SimpleHTTPRequestHandler
from socketserver import TCPServer

with TCPServer(("", 8080), partial(SimpleHTTPRequestHandler, directory=".")) as httpd:
    httpd.serve_forever()

有没有使用asyncio 的简单方法来做到这一点？

Answer 1

没有简单的方法。 asyncio 旨在成为框架。

如果你必须这样做，这是我花了 4 个小时研究的东西 - 我从来不知道在不使用 3rd 方库时我必须研究这么多东西，预计可能质量差的代码，因为我以前从未处理过原始请求/响应。

代码

import asyncio
import pathlib
import textwrap
from pprint import pprint


ROOT = pathlib.Path(__file__).parent


async def read_all(r: asyncio.StreamReader):
    output = ''

    while recv := await r.read(1024):
        output += recv.decode("utf8")

        if len(recv) < 1024:
            break

    return output


def parse_req(req: str):
    line_iter = iter(req.rstrip().splitlines())

    # 1st line is head, all other can just be put into dict
    method, dir_, http_ver = next(line_iter).split()

    req_dict = {
        'Method': method,
        'Directory': "" if dir_ == "/" else dir_,
        'HTTP': http_ver
    }

    for line in line_iter:
        key, val = line.split(": ")
        req_dict[key] = val

    return req_dict


def file_list_html_gen(path: pathlib.Path):

    for sub_path in path.iterdir():

        if path == ROOT:
            relative = '/'
        else:
            relative = '/' + str(path.relative_to(ROOT)).strip('./') + '/'

        file_name = sub_path.name

        if sub_path.is_dir():
            yield f'<a href="{relative}{file_name}">{file_name}</a>'
        else:
            yield f'<a href="{relative}{file_name}" download="{file_name}">{file_name}</a>'


def create_resp(req_dict: dict):

    # reject user when POST or other stuffs are used
    if req_dict['Method'] != 'GET':
        return f"{req_dict['HTTP']} 405 Method Not Allowed\r\nConnection: close\r\n\r\n", None

    # reject user when accessing not existing directory
    dir_ = ROOT.joinpath(req_dict["Directory"][1:])

    if not dir_.exists():
        return f"{req_dict['HTTP']} 404 Not Found\r\nConnection: close\r\n\r\n", None

    # else build html for directory
    if dir_.is_dir():
        try:
            parent_str = str(dir_.parent.relative_to(ROOT)).strip('./')

        except ValueError:
            parent_str = ""

        parent_dir = f'<a href="/{parent_str}">Go Up</a><br>'
        path_list = parent_dir + '<br>'.join(file_list_html_gen(dir_))

        resp = f"{req_dict['HTTP']} 200 OK\r\n" \
               f"Content-Type: text/html\r\n" \
               f"Content-Length: {len(path_list.encode('utf8'))}\r\n" \
               f"Connection: close\r\n\r\n" \
               f"{path_list}"

        attach = None

    else:
        # else send file
        attach = dir_.read_bytes()

        resp = f"{req_dict['HTTP']} 200 OK\r\n" \
               f"Content-Type: application/octet-stream\r\n" \
               f"Content-Length: {len(attach)}\r\n" \
               f"Connection: close\r\n\r\n"

    return resp, attach


async def tcp_handler(r: asyncio.StreamReader, w: asyncio.StreamWriter):
    # Receive
    print("\nReceiving")

    parsed = parse_req(await read_all(r))
    pprint(parsed)

    print("Received")

    resp, file = create_resp(parsed)

    # Respond
    print("\nResponding", textwrap.indent(resp, " "), sep='\n')
    w.write(resp.encode('utf8'))
    if file:
        print("File length:", len(file))
        w.write(file)
    await w.drain()
    w.close()

    print("Response sent")


async def serve_files():
    server = await asyncio.start_server(tcp_handler, '127.0.0.1', 80)

    print("Started")

    async with server:
        await server.serve_forever()


if __name__ == '__main__':
    asyncio.run(serve_files())

---

输出

Started

Receiving
{'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
 'Accept-Encoding': 'gzip, deflate, br',
 'Accept-Language': 'ko-KR,ko;q=0.9,en-US;q=0.8,en;q=0.7,zh-TW;q=0.6,zh;q=0.5',
 'Cache-Control': 'no-cache',
 'Connection': 'keep-alive',
 'Directory': '',
 'HTTP': 'HTTP/1.1',
 'Host': '127.0.0.1',
 'Method': 'GET',
 'Pragma': 'no-cache',
 'Referer': 'http://127.0.0.1/Tools',
 'Sec-Fetch-Dest': 'document',
 'Sec-Fetch-Mode': 'navigate',
 'Sec-Fetch-Site': 'same-origin',
 'Sec-Fetch-User': '?1',
 'Upgrade-Insecure-Requests': '1',
 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
               '(KHTML, like Gecko) Chrome/96.0.4664.110 Safari/537.36',
 'sec-ch-ua': '" Not A;Brand";v="99", "Chromium";v="96", "Google '
              'Chrome";v="96"',
 'sec-ch-ua-mobile': '?0',
 'sec-ch-ua-platform': '"Windows"',
 'sec-gpc': '1'}
Received

Responding
 HTTP/1.1 200 OK
 Content-Type: text/html
 Content-Length: 639
 Connection: close

 <a href="/">Go Up</a><br><a href="/cc.py" download="cc.py">cc.py</a><br><a href="/height_data.txt" download="height_data.txt">height_data.txt</a><br><a hr
ef="/main.py" download="main.py">main.py</a><br><a href="/sample.json" download="sample.json">sample.json</a><br><a href="/scratch.kv" download="scratch.kv
">scratch.kv</a><br><a href="/scratch.md" download="scratch.md">scratch.md</a><br><a href="/scratch.py" download="scratch.py">scratch.py</a><br><a href="/s
cratch3.py" download="scratch3.py">scratch3.py</a><br><a href="/test.py" download="test.py">test.py</a><br><a href="/Tools">Tools</a><br><a href="/__pycach
e__">__pycache__</a>
Response sent

---

您可以像任何其他生成的 html 一样导航、下载，但视觉效果不佳并且（可能）无法从服务器端发送大文件。

我会说确实有可能像这样 - 但我不认为纯粹是不值得的麻烦。我们不必重新发明轮子吗？