Litestar 高级主题实战指南同步与异步执行、显式参数声明与生产部署【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestarLitestar 是一个轻量、灵活且可扩展的 ASGI 框架。本文是 docs/topics/index.rst 所收录的三篇高级技术主题的系统化讲解同步与异步执行模型阻塞/非阻塞、I/O 密集与 CPU 密集任务、线程池的取舍、显式参数声明为什么 Litestar 要求用标记显式声明 handler 与依赖参数来源以及它如何帮你提前发现错误以及生产环境部署手动 ASGI 服务器、Docker、Supervisor 三种方案的完整实操。读完本文你将掌握如何为并发场景选择正确的执行模式、如何写出信息密度更高且不易出错的参数声明以及如何把应用稳定地部署到 Linux 生产环境。一、同步与异步Litestar 的三种执行模式Litestar 几乎在一切可行的位置同时支持同步与异步可调用对象。概括来说框架支持三种执行模式直接运行异步可调用对象async callable直接运行同步可调用对象sync callable在线程池中运行同步可调用对象thread pool理解这三种模式的区别是写出高性能 Litestar 应用的基础。本节内容对应 docs/topics/sync-vs-async.rst。阻塞与非阻塞Blocking and non-blocking在异步编程语境中阻塞与非阻塞通常用来描述一个函数的某种品质是否阻塞了执行流。需要特别强调的是异步函数并不是天然非阻塞的。它真正做的是让程序员精确控制在何处让出控制权并交还给事件循环从而允许其他异步任务运行这一行为通常由await关键字标记。这一点至关重要——一个从不调用await、却执行了大量计算任务的异步函数会像同步函数一样在其整个运行期间阻塞主线程。更关键的是异步函数内部两次等待之间发生的一切同样会被阻塞。从技术上讲Python 中并不存在真正非阻塞的函数因为异步函数也只在每个await处解锁。因此通常意义上的阻塞指的是长时间阻塞执行流的可调用对象。关于await的辨析从执行流的角度看await本身也是阻塞的——在 awaitable 被解析之前执行不会越过它。但由于此时程序中其他部分更准确地说是那些在某个await处让出控制权、且该等待已完成的协程可以继续推进因此这不被视为阻塞通常被称为等待waiting。从源码实现看Litestar 通过 litestar/utils/sync.py 中的ensure_async_callable与AsyncCallable来统一处理同步/异步可调用对象若传入的函数本身是异步的直接原样返回否则包装为AsyncCallable在调用时通过sync_to_thread放到线程池中执行。I/O 密集与 CPU 密集I/O bound vs. CPU bound判断一个函数是否应该异步关键看它为什么会阻塞。阻塞操作大致分为两类I/O 密集I/O bound对文件系统或网络的调用是典型的 I/O 密集阻塞。它们的执行速度主要受限于外部资源完成操作所需的时间因此绑定在该资源上。这类操作非常适合异步大部分时间都花在等待上其他任务可以在这段时间内并行等待从而显著降低整体运行时间。CPU 密集CPU bound不等待 I/O 的操作通常可以视为 CPU 密集。由于它们不等待外部资源执行速度绑定在 CPU 上。这类任务无法像 I/O 密集任务那样从异步执行中获益因为它们不会花费大量时间在等待上。异步 CPU 密集任务在某些情况下可以通过引入事件循环切换任务的让步点来让 CPU 密集任务异步化。一个经典例子是在内层循环的每次迭代开头await asyncio.sleep(0)。这种技巧主要适用于长时运行的任务其中每个单独的步骤不会长时间阻塞。何时使用异步函数异步函数应在能从并发执行中获益时使用即它们自身执行异步操作如调用其他异步函数、异步迭代且不执行任何阻塞操作如调用 I/O 密集的同步函数。为什么不默认全部用 async一个常见的诱惑是除非同步执行阻塞操作否则函数都应该默认 async——事实并非如此。async 本身有额外开销虽然极小但在某些场景下不可忽视。一个执行非阻塞操作的同步函数性能会优于做同样事情的异步函数。何时使用同步函数作为上一节的推论同步函数应用于非 I/O 密集的任务。同步执行模型的开销最小在没有使用任何异步功能的情况下应当优先选择同步写法。何时使用线程池如果函数执行计算密集或 I/O 密集操作且无法用异步等价物替代可以使用sync_to_threadTrue将其放入线程池运行get(/) def sync_hello_world() - dict[str, object]: return {hello: world} get(/sync-to-thread, sync_to_threadTrue) def blocking_hello_world() - dict[str, object]: # 计算密集或阻塞 I/O 操作 return {hello: world}sync_to_thread是各 HTTP handler 装饰器get/post等共有的参数定义见 litestar/handlers/http_handlers/decorators.py最终在 litestar/handlers/http_handlers/base.py 中被解析当该参数为None默认且函数是同步函数时会触发隐式线程池切换并发出警告为True时强制放入线程池为False时直接在事件循环中运行。为什么不让同步函数默认进线程池线程池的运行开销非常高会大幅降低应用性能。它的使用应严格限制在确有必要的情形。局限虽然把 CPU 密集函数放进线程池可以让它非阻塞地运行但并不会加速其执行。需要定期执行的计算密集任务应卸载到独立进程中以利用多个 CPU 核心——例如通过anyio.to_process.run_sync实现。从底层实现看litestar/concurrency.py 中的sync_to_thread会根据当前事件循环选择执行后端在 asyncio 环境下通过asyncio.get_running_loop().run_in_executor(...)运行并借助contextvars.copy_context()保留上下文在 trio 环境下通过trio.to_thread.run_sync(...)运行。你还可以通过set_asyncio_executor/set_trio_capacity_limiter自定义执行器与容量限制器。关于执行模式的警告由于同步函数可能阻塞Litestar 会在可能阻塞主事件循环、影响应用性能的位置对其使用发出警告。如果某个同步函数确实非阻塞设置sync_to_threadFalse即可告知 Litestar 将其视为非阻塞函数。引入这一警告的目的是强制开发者对是否在线程池中运行该函数做出明确决策从而避免无意中使用阻塞函数。全局禁用警告的方式是设置环境变量export LITESTAR_WARN_IMPLICIT_SYNC_TO_THREAD0该开关的实现位于 litestar/utils/warnings.py 中的warn_implicit_sync_to_thread默认开启通过envflag(LITESTAR_WARN_IMPLICIT_SYNC_TO_THREAD, defaultTrue)判断是否输出警告。相关测试可参考 tests/unit/test_utils 中的用例。二、显式参数声明Litestar 对易读优于易写的设计实践Litestar 拥抱并强制要求对 handler / 依赖参数进行显式声明。本节内容对应 docs/topics/explicit_declarations.rst旨在解释这一设计决策背后的动机并回应关于其冗长的常见疑问。什么是显式参数声明简单来说在 Litestar 中显式参数声明是指任何用标记marker如FromQuery标注的 handler 或依赖函数参数from litestar import get from litestar.params import FromQuery get(/) async def handler(name: FromQuery[str]) - str: return nameFromQuery在 litestar/params.py 中被定义为Annotated[T, QueryParameter()]的类型别名即一个携带查询参数元数据的注解。完整的参数使用方式可参考 docs/usage/routing/parameters.rst。历史背景从隐式推断到显式声明早期版本的 Litestar 允许隐式参数即通过推断决定参数来源主要适用于路径、查询和依赖参数。例如get(/) async def handler(message: str) - str: return message这里因为message未被其他来源消费它会隐式地成为一个查询参数。这种设计受到 FastAPI、DRF 等框架的影响旨在提供书写便利。然而问题很快显现如果同一个路由 handler 被放进一个包含名为message的路径参数的路由器中路径参数会胜出message就不再是查询参数了。这意味着推断出的参数来源是上下文相关的context dependent。当然隐式方式也有其优点比如可以将其视为通过依赖注入实现控制反转的一种实际应用被调用方声明参数时无需知道调用它的上下文只需声明依赖由调用方决定如何提供。不过这一优点在显式声明下依然成立被调用方只是额外声明了它需要的参数种类——而这在大多数情况下并非纯装饰性的差异而是携带了实际语义一个x-api-key请求头与同名查询参数绝不等价。动机一易读优于易写为每个声明多写几个符号确实意味着更多打字量无可辩驳。但既然代码被阅读的次数远多于被编写的次数参考资料见《从代码复杂度度量到程序理解》库的首要目标应当是易于理解。因此 Litestar 在设计显式参数声明时追求的目标是任何读者即使不熟悉该框架也能仅凭声明本身或许再看一眼 docstring理解其含义。易写但不易读的版本get(/) async def handler( data: dict[str, str], limit: int, page: Annotated[int, Parameter(ge1)], db_session: AsyncSession, ) - None: ...这很容易打出来但读者能获得的信息很少data和其他参数从哪来可以从limit、page推断它们大概是查询参数但正如前文历史背景所述这也可能是上下文相关的。易读的版本get(/) async def handler( data: JSONBody[dict[str, str]], limit: Annotated[int, QueryParameter()], page: Annotated[int, QueryParameter(ge1)], db_session: NamedDependency[AsyncSession], ) - None: ...这确实冗长得多但信息量也大得多data: JSONBody[dict[str, str]]告诉我们这是请求体request body且请求体是 JSON 格式limit: Annotated[int, QueryParameter()]这就是一个查询参数page: Annotated[int, QueryParameter(ge1)]这是另一个查询参数且最小值为 1db_session: NamedDependency[AsyncSession]这是一个按名称注入的依赖。JSONBody与QueryParameter均定义于 litestar/params.pyJSONBody是Annotated[T, BodyKwarg(media_typeRequestEncodingType.JSON)]的别名NamedDependency则定义于 litestar/di.py其 docstring 明确说明函数参数的名称将作为要注入的依赖的名称。动机二避免错误——提前失败显式方案带来一个显著好处能够尽早失败fail early。如果某个参数没有声明来源Litestar 可以立即报错避免耗时的运行时调试。以基于推断的方式为例一个常见且有时令人困惑的错误是missing required query parameter xxx当依赖未被提供时。原因在于参数被解释为依赖还是查询参数取决于是否存在对应的依赖提供者。如果不存在提供者参数会被解释为查询参数若没有默认值则被标记为必填get(/) async def get_page(db_session: AsyncSession) - None: ... router_a Router(/a, [get_page], dependencies{db_session: provide_session}) router_b Router(/b, [get_page])调用/a会正常工作但调用/b会返回400 - Bad Request提示missing required query parameter db_session——而实际上db_session本应是一个依赖。改用显式声明后get(/) async def get_page(db_session: NamedDependency[AsyncSession]) - None: ... router_a Router(/a, [get_page], dependencies{db_session: provide_session}) router_b Router(/b, [get_page])这些情况要么在启动时被检测出来要么抛出一个具体且正确的错误此时会是500 - Internal Server Error并在 stderr 中给出更有帮助的错误信息Explicit dependency db_session for get_page has no default value, or provided dependency.显式依赖 db_session 对于 get_page 没有默认值或已提供的依赖。动机三文档化意图使用隐式声明时意图也是隐式地被文档化——即根本没有被文档化。未被代码表达的意图通常只有两种归宿写进注释或者最终被遗忘。通过显式声明意图始终编码在使用现场无需额外维护一份容易过期的注释。三、生产环境部署三种主流方案docs/topics/deployment/index.rst见 docs/topics/deployment/index.rst汇集了将 Litestar 应用部署到各类平台与环境的文章包括systemd、Docker、Kubernetes、Serverless 等。本节详细讲解其中已完善的三篇手动 ASGI 服务器、Docker 与 Supervisor。方案一手动使用 ASGI 服务器ASGIAsynchronous Server Gateway Interface旨在为 Litestar 这类异步 Python Web 框架与异步 Web 服务器之间提供标准接口。市面上有多款流行的 ASGI 服务器可供选择。适用场景手动使用 ASGI 服务器运行应用通常只适合开发与测试环境。生产负载一般推荐放入容器化环境如 Docker 或 Kubernetes或通过进程控制系统如 Supervisor 或systemd管理。备选方案一览systemd集成在许多 Linux 发行版中的系统与服务管理器用于管理系统进程官方文档即将推出Supervisor进程控制系统可自动启动、停止和重启进程自带 Web UIDocker适合容器化环境提供隔离性与可伸缩性。ASGI 服务器选型服务器特点协议支持Uvicorn主流 ASGI 服务器HTTP/1.1、WebSocketHypercorn最初源自 QuartHTTP/1.1、HTTP/2、WebSocketDaphne最初为 Django Channels 开发HTTP/1.1、HTTP/2、WebSocketGranian基于 Rust 实现HTTP/1.1、HTTP/2、WebSocket安装pip install uvicorn # Uvicorn pip install hypercorn # Hypercorn pip install daphne # Daphne pip install granian # Granian运行假设应用按快速入门中的 Minimal Example 定义于app.py应用实例名为appuvicorn app:app # INFO: Waiting for application startup. # INFO: Application startup complete. # INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRLC to quit) hypercorn app:app # [2023-11-12 23:31:26 -0800] [16748] [INFO] Running on http://127.0.0.1:8000 (CTRL C to quit) daphne app:app # INFO - 2023-11-12 23:31:51,571 - daphne.cli - cli - Starting server at tcp:port8000:interface127.0.0.1 # INFO - 2023-11-12 23:31:51,572 - daphne.server - server - Listening on TCP address 127.0.0.1:8000 granian --interface asgi app:app # [INFO] Starting granian # [INFO] Listening at: 127.0.0.1:8000关于 Gunicorn Uvicorn 的弃用说明GunicornUvicorn 组合模式在 ASGI 部署中已被视为 legacy因为 Uvicorn 0.30.0 已内置原生 worker 管理。Uvicorn 新增的多进程管理器旨在完全取代 Gunicorn实现细节见 Uvicorn 的 PR #2183。新部署请直接使用 Uvicorn。方案二Docker 容器化部署Docker 是一个容器化平台可将应用及其全部依赖打包在一起无论宿主系统的配置与依赖如何都能创建一致的运行环境尤其有助于避免依赖冲突。本指南使用 Docker 官方 Python 容器 作为基础镜像。适用场景需要一致的隔离环境Isolation、按需伸缩Scalability、跨环境一致性运行Portability、微服务架构Microservices、CI/CD 流水线Dependency Management 与自动化部署以及依赖打包管理。前提系统已安装并运行 Docker项目目录中包含以下文件requirements.txtlitestar[standard]2.4.0,3.0.0app.pyMinimal Litestar application. from asyncio import sleep from typing import Any, Dict from litestar import Litestar, get get(/) async def async_hello_world() - Dict[str, Any]: Route Handler that outputs hello world. await sleep(0.1) return {hello: world} get(/sync, sync_to_threadFalse) def sync_hello_world() - Dict[str, Any]: Route Handler that outputs hello world. return {hello: world} app Litestar(route_handlers[sync_hello_world, async_hello_world])注意这个示例同时演示了前文所述的两种执行模式异步 handlerasync_hello_world内部await sleep(0.1)模拟异步等待与显式声明为sync_to_threadFalse的同步 handlersync_hello_world。示例 Dockerfile# Set the base image using Python 3.12 and Debian Bookworm FROM python:3.12-slim-bookworm # Set the working directory to /app WORKDIR /app # Copy only the necessary files to the working directory COPY . /app # Install the requirements RUN pip install --no-cache-dir --upgrade -r /app/requirements.txt # Expose the port the app runs on EXPOSE 80 # Run the app with the Litestar CLI CMD [litestar, run, --host, 0.0.0.0, --port, 80]该 Dockerfile 将本地项目目录复制到容器内的/app并通过litestar run命令其底层使用uvicorn运行应用。uvicorn由litestar[standard]extra 提供已在requirements.txt中安装。当然你也可以选择直接用 ASGI 服务器 启动应用。常用 Docker 命令# Build the container docker build -t exampleapp . # Run the container docker run -d -p 80:80 --name exampleapp exampleapp # Stop the container docker stop exampleapp # Start the container docker start exampleapp # Remove the container docker rm exampleappDocker ComposeCompose 是定义和运行多容器 Docker 应用的工具。若要将容器纳入 Compose 编排可使用如下docker-compose.ymlversion: 3.9 services: exampleapp: build: context: ./ dockerfile: Dockerfile container_name: exampleapp depends_on: - database ports: - 80:80 environment: - DB_HOSTdatabase - DB_PORT5432 - DB_USERlitestar - DB_PASSr0cks - DB_NAMEexampleapp database: image: postgres:latest container_name: exampledb environment: POSTGRES_USER: exampleuser POSTGRES_PASSWORD: examplepass POSTGRES_DB: exampledb ports: - 5432:5432 volumes: # Note that image versions previous to 18 used /var/lib/postgresql/data - db_data:/var/lib/postgresql volumes: db_data:该 Compose 文件定义了两个服务exampleapp与database。exampleapp由当前目录下的 Dockerfile 构建并暴露 80 端口database使用官方 PostgreSQL 镜像并暴露 5432 端口exampleapp通过depends_on依赖database因此数据库会先于应用启动并通过环境变量注入数据库连接信息供应用使用。常用 Compose 命令# Build the containers docker compose build # Run the containers docker compose up # Run the containers in the background docker compose up -d # Stop the containers docker compose down方案三Supervisor (Linux) 进程管理supervisor是 Linux 上的进程控制系统允许在类 UNIX 操作系统上监控和控制多个进程。它特别适合管理需要持续运行的后台进程以及需要监控和控制的后台进程。适用场景需要持续在线与稳健进程控制的 Python Web 应用尤其是需要完善的进程管理、监控、日志管理以及对进程启动/停止/重启的定制化控制时。详细信息可参考 Supervisor 官方文档。备选方案Docker容器化隔离与伸缩、systemd系统与服务管理器、手动 ASGI 服务器直接控制。配置Supervisor 使用配置文件定义服务详见 Supervisor 配置文档。将以下配置放到/etc/supervisor/conf.d/exampleapp.conf[program:exampleapp] # Defines the service name. directory/opt/exampleapp/src # Specifies the directory where the service should run. command/opt/exampleapp/venv/bin/litestar app.py # Specifies the command to run the service. redirect_stderrtrue # Redirects stderr to stdout. stdout_logfile/var/log/exampleapp.log # Specifies the log file to write to. stdout_logfile_backups10 # Specifies the number of backups to keep. autostarttrue # Specifies that the service should start automatically when supervisor starts. autorestarttrue # Specifies that the service should restart automatically if it exits unexpectedly.创建配置文件后需要重新加载 Supervisor 配置以载入新服务# Reload supervisor config sudo supervisorctl reread sudo supervisorctl update启动与验证# Start/Stop/Restart/Status sudo supervisorctl start exampleapp sudo supervisorctl stop exampleapp sudo supervisorctl restart exampleapp sudo supervisorctl status exampleapp # View logs sudo supervisorctl tail -f exampleapp完成以上步骤后即可启动应用先执行sudo supervisorctl start exampleapp再通过sudo supervisorctl status exampleapp检查是否正常默认监听http://0.0.0.0:8000。以下小节是可选的便捷化建议。建议一命令别名。创建别名文件/etc/profile.d/exampleapp.sh即可用exampleapp start替代sudo supervisorctl start exampleappexampleapp() { case $1 in start) echo Starting exampleapp... sudo supervisorctl start exampleapp ;; stop) echo Stopping exampleapp... sudo supervisorctl stop exampleapp ;; restart) echo Restarting exampleapp... sudo supervisorctl restart exampleapp ;; status) echo Checking status of exampleapp... sudo supervisorctl status exampleapp ;; watch) echo Tailing logs for exampleapp... sudo supervisorctl tail -f exampleapp ;; help) cat EOF Available options: exampleapp start - Start the exampleapp service exampleapp stop - Stop the exampleapp service exampleapp restart - Restart the exampleapp service exampleapp status - Check the status of the exampleapp service exampleapp watch - Tail the logs for the exampleapp service EOF ;; *) echo Unknown command: $1 echo Use exampleapp help for a list of available commands. ;; esac }无需重启会话即可生效source /etc/profile.d/exampleapp.sh。其中watch命令可实时监控应用输出。建议二一键更新脚本。在exampleapp函数中扩展update命令实现完整的应用更新流程exampleapp() { case $1 in # ... other cases ... # update) echo Updating exampleapp... # Stop the service echo Stopping service... sudo supervisorctl stop exampleapp # Update application files echo Pulling latest changes from repository... cd /opt/exampleapp git fetch --all git reset --hard origin/master # Update Supervisor configuration and alias echo Updating Supervisor and shell configurations... sudo ln -sf /opt/exampleapp/server/service.conf /etc/supervisor/conf.d/exampleapp.conf sudo ln -sf /opt/exampleapp/server/alias.sh /etc/profile.d/exampleapp.sh source /etc/profile.d/exampleapp.sh # Update Supervisor to apply new configurations echo Reloading Supervisor configuration... sudo supervisorctl reread sudo supervisorctl update # Update Python dependencies using requirements.txt # Here you could replace with poetry, pdm, etc., alleviating the need for # a requirements.txt file and virtual environment activation. source venv/bin/activate echo Installing updated dependencies... python3 -m pip install -r requirements.txt deactivate # ... other update processes like docs building, cleanup, etc. ... # echo Update process complete. # Prompt to start the service read -p Start the service? (y/n) -n 1 -r echo if [[ $REPLY ~ ^[Yy]$ ]] then echo Starting service... sudo supervisorctl start exampleapp fi ;; # ... # esac }该更新流程包含以下步骤停止服务改动前安全停服→Git 操作从仓库拉取最新代码→配置符号链接更新 Supervisor 配置与 shell 别名→重载 Supervisor应用新的配置→依赖更新按requirements.txt或锁文件安装/更新依赖→用户确认询问是否立即启动服务。之后执行exampleapp update即可一键完成从拉码、更新依赖到重载配置的完整部署周期。总结Litestar 的三篇高级主题分别回答了三个关键问题执行模型上框架提供直接异步、直接同步、线程池三种模式核心原则是异步用于能并发受益的场景、同步用于零开销场景、线程池用于阻塞且无法异步化的场景并可通过sync_to_thread显式控制与LITESTAR_WARN_IMPLICIT_SYNC_TO_THREAD控制警告参数声明上显式标记FromQuery、QueryParameter、JSONBody、NamedDependency等把易读优于易写落到实处让意图在代码现场自文档化并在启动阶段提前暴露缺失依赖等错误部署上从开发期的 Uvicorn/Hypercorn/Daphne/Granian 手动运行到生产期的 Docker 容器化含 Compose 多服务编排与 Supervisor 进程守护含别名与一键更新脚本构成了从开发到上线的完整路径。相关原始资料可在 docs/topics 目录下继续深入阅读核心实现可对照 litestar/concurrency.py、litestar/utils/sync.py、litestar/params.py 与 litestar/di.py。【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考