Reflex API Transformer 实战指南用 FastAPI 与 Starlette 扩展 Python Web 后端【免费下载链接】reflex️ Web apps in pure Python 项目地址: https://gitcode.com/GitHub_Trending/re/reflexReflex 在渲染前端应用的同时底层由 FastAPI 承载后端服务。API Transformer 是rx.App提供的一项扩展机制允许你在不破坏 Reflex 运行时事件、上传、心跳等的前提下将已有的 FastAPI/Starlette 应用挂载进来、为其补充自定义 API 路由或以中间件形式包装整个 ASGI 应用。读完本文你将掌握api_transformer参数的三种用法实例挂载、可调用转换、多转换器链式组合理解它在 app.py 中的实际执行顺序并明确哪些后端路由是运行时保留、不可随意覆盖的。认识 API Transformer一个应用、两层后端Reflex 应用的运行时结构是「前端 后端」前端由 React 编译产出后端则是一个 FastAPI基于 Starlette应用负责页面预渲染、事件处理、状态同步与文件上传等职责。这意味着 Reflex 应用天然就是一个 ASGI 应用天然具备扩展 HTTP 接口的能力。API Transformer 的定位正是在 Reflex 内部 ASGI 应用正式对外服务之前对它做一次或多次变换官方文档 docs/api-routes/overview.md 明确列出了它的三个用途将已有的 FastAPI 或 Starlette 应用与 Reflex 应用集成对 ASGI 应用应用中间件或做整体变换为 Reflex 应用扩展额外的 API 端点。典型应用场景包括为 Reflex 应用补充一套独立的业务 REST API例如移动端/第三方消费的接口、接入现有的认证体系OAuth2、JWT、复用团队已有的 Starlette 中间件或统一为后端加上 CORS、日志、限流等横切能力。快速上手把 FastAPI 应用挂进 Reflex用法非常直接初始化rx.App时传入api_transformer参数即可。下面是一个最小示例——先创建一个 FastAPI 实例并定义路由再把它作为 transformer 传入import reflex as rx from fastapi import FastAPI, Depends from fastapi.security import OAuth2PasswordBearer # Create a FastAPI app fastapi_app FastAPI(titleMy API) # Add routes to the FastAPI app fastapi_app.get(/api/items) async def get_items(): return dict(items[Item1, Item2, Item3]) # Create a Reflex app with the FastAPI app as the API transformer app rx.App(api_transformerfastapi_app)启动 Reflex 应用后/api/items即可直接访问而 Reflex 自身的页面路由与事件通道照常工作。从源码看api_transformer是rx.App的一个公开字段其类型定义为api_transformer: ( Sequence[Callable[[ASGIApp], ASGIApp] | Starlette] | Callable[[ASGIApp], ASGIApp] | Starlette | None ) None见 reflex/app.py字段注释说明它是在运行前应用到后端 ASGI 应用上的一次或多次变换——挂载 FastAPI/Starlette 应用或用 ASGI 中间件包装它reflex/app.py。三种 Transformer 类型详解api_transformer可接受三类值你可以根据需求选择也可以组合使用。类型一FastAPI / Starlette 实例挂载式集成传入一个 FastAPI 或 Starlette 实例时Reflex 会把它的内部 API 挂载到你的应用上从而让你既能使用 Reflex 的运行时又能自由定义额外路由。文档给出的带认证的完整示例import reflex as rx from fastapi import FastAPI, Depends from fastapi.security import OAuth2PasswordBearer # Create a FastAPI app with authentication fastapi_app FastAPI(titleSecure API) oauth2_scheme OAuth2PasswordBearer(tokenUrltoken) # Add a protected route fastapi_app.get(/api/protected) async def protected_route(token: str Depends(oauth2_scheme)): return dict(messageThis is a protected endpoint) # Create a token endpoint fastapi_app.post(/token) async def login(username: str, password: str): # In a real app, you would validate credentials if username user and password password: return dict(access_tokenexample_token, token_typebearer) return dict(errorInvalid credentials) # Create a Reflex app with the FastAPI app as the API transformer app rx.App(api_transformerfastapi_app)这个例子的价值在于Reflex 的前端可以直接调用/api/protected等受保护接口而 FastAPI 的Depends/OAuth2PasswordBearer认证链路完全独立于 Reflex 状态系统实现了前端框架 后端 API 网关的清晰分层。类型二可调用转换器中间件式包装也可以传入一个接收 ASGIApp、返回 ASGIApp的可调用对象从而对 Reflex 后端做整体包装。文档以 CORS 中间件为例import reflex as rx from starlette.middleware.cors import CORSMiddleware # Create a transformer function that returns a transformed ASGI app def add_cors_middleware(app): # Wrap the app with CORS middleware and return the wrapped app return CORSMiddleware( appapp, allow_origins[https://example.com], allow_methods[*], allow_headers[*], ) # Create a Reflex app with the transformer app rx.App(api_transformeradd_cors_middleware)只要签名满足Callable[[ASGIApp], ASGIApp]任何中间件或装饰器式包装都能这样接入包括自定义的请求日志、认证校验、请求 ID 注入等。类型三多个转换器组合链式应用将多个 transformer 放入列表/元组Reflex 会按顺序依次应用。文档示例同时组合了一个 FastAPI 实例和一个自定义日志中间件import reflex as rx from fastapi import FastAPI from starlette.middleware import Middleware from starlette.middleware.cors import CORSMiddleware # Create a FastAPI app fastapi_app FastAPI(titleMy API) # Add routes to the FastAPI app fastapi_app.get(/api/items) async def get_items(): return dict(items[Item1, Item2, Item3]) # Create a transformer function def add_logging_middleware(app): # This is a simple example middleware that logs requests async def middleware(scope, receive, send): # Log the request path path scope[path] print(Request:, path) await app(scope, receive, send) return middleware # Create a Reflex app with multiple transformers app rx.App(api_transformer[fastapi_app, add_logging_middleware])注意顺序语义序列中的元素从左到右依次生效。由于列表里既有实例又有可调用对象两者会被区分对待见下文的源码执行流程最终形成的是一条完整的 ASGI 处理链。源码视角api_transformer 的执行流程要理解这三种类型的差异需要看App.__call__中对api_transformer的实际处理逻辑reflex/app.pyif self.api_transformer is not None: api_transformers: Sequence[Starlette | Callable[[ASGIApp], ASGIApp]] ( [self.api_transformer] if not isinstance(self.api_transformer, Sequence) else self.api_transformer ) for api_transformer in api_transformers: if isinstance(api_transformer, Starlette): # Mount the api to the starlette app. App._add_cors(api_transformer) api_transformer.mount(, asgi_app) asgi_app api_transformer else: # Transform the asgi app. asgi_app api_transformer(asgi_app) top_asgi_app Starlette(lifespanself._run_lifespan_tasks) # Make sure Reflex contexts are attached for each request. top_asgi_app.mount( , self._context_middleware(asgi_app), ) App._add_cors(top_asgi_app) return top_asgi_app几个关键事实均由这段源码直接印证先规范化再逐个应用传入单个 transformer 时会被包装成单元素序列因此三种类型的处理路径是统一的。实例走挂载可调用走包装对Starlette子类FastAPI 继承自 Starlette实例Reflex 调用api_transformer.mount(, asgi_app)把 Reflex 内部 API 挂到你的应用根路径下随后以你的应用作为新的一层对可调用对象则直接asgi_app api_transformer(asgi_app)完成包装。这也解释了为何传入 FastAPI 实例时你定义的路由能与 Reflex 路由共存。自动附加 CORSApp._add_cors(api_transformer)reflex/app.py会在你的 FastAPI 实例上同样注册CORSMiddlewareallow_credentialsTrue意味着通过 transformer 挂载进来的 API 也默认获得与 Reflex 后端一致的跨域策略。上下文中间件与 lifespan 由最外层保证处理完所有 transformer 后Reflex 会创建一个带_run_lifespan_taskslifespan 的顶层Starlette并通过_context_middleware为每个请求设置 Reflex 上下文reflex/app.py挂载整个变换后的应用最后再对顶层应用补一次 CORS。也就是说无论你怎么包装事件处理所需的请求上下文都不会丢。保留路由不要覆盖的运行时端点Reflex 后端有一些专供运行时使用的保留路由它们由App._add_default_endpoints与App._add_optional_endpoints在内部注册reflex/app.py并会挂载到内部self._api上随 transformer 一起对外暴露。覆盖它们会破坏运行时通信除非你完全清楚后果。路由作用注意事项/ping/后端健康检查期望返回pong源码中ping端点返回JSONResponse(pong)见 reflex/app.py/_event前端事件通知后端的事件通道Socket.IO 挂载点覆盖该路由会破坏事件通信导致页面状态无法更新/_uploadrx.upload()文件上传入口供rx.upload_files(...)、rx.upload_files_chunk(...)使用覆盖后上传功能失效相关实现见 reflex/app.py从源码可以看到_add_default_endpoints会注册ping与health两个 GET 路由_add_optional_endpoints则在检测到上传组件被使用时注册_upload的 POST 路由并挂载一个静态文件服务用于访问已上传文件而_event路由在App.__init__阶段就通过config.prepend_backend_path(str(constants.Endpoint.EVENT))将 Socket.IO 应用挂载到内部 API 上reflex/app.py。由于这些端点注册发生在 transformer 应用之前它们会被完整保留——但这也意味着如果你在自己的 FastAPI 实例上定义了同名路由就可能与运行时端点产生冲突这正是文档强调不要覆盖的原因。实战验证测试用例给出的集成范式仓库集成测试提供了api_transformer与真实后端协作的参考实现tests/integration/test_lifespan.py 中LifespanApp以rx.App(api_transformerFastAPI() if mount_api_transformer else None)构造应用并用参数化 fixtureno_api_transformer/mount_api_transformer验证无论是否挂载 transformerlifespan 任务与请求上下文都能正常工作。tests/integration/test_linked_state.py 展示了rx.App(api_transformerapi)与动态路由/room/[room]同时使用的写法印证transformer 挂载 API 与 Reflex 前端功能互不干扰。tests/units/docgen/test_class_and_component.py 则从文档生成角度验证了api_transformer字段的类型展示清晰可读。这些测试同时说明了最佳实践把自定义 API 与 Reflex 页面分离组织——页面负责交互transformer 挂载的 FastAPI 负责对外服务二者通过同一 ASGI 进程对外暴露部署时无需额外进程。注意事项与最佳实践保留路由优先让位不要在自定义 FastAPI 实例中定义/ping/、/_event、/_upload同名路由确需调整时优先在应用外层如 Caddy/Nginx做路径改写而不是覆盖运行时端点。理解应用顺序多个 transformer 按列表顺序生效实例型做挂载、可调用型做包装需要精确控制中间件执行次序时将它们组织成明确的顺序序列。CORS 自动附加挂载进来的 FastAPI 实例会被自动加上 Reflex 的 CORS 中间件若你的 API 需要更细粒度的跨域策略可在自己的实例上显式配置覆盖。生命周期一致transformer 不改变顶层应用的 lifespan 管理Reflex 的 lifespan 任务tests/integration/test_lifespan.py 有完整覆盖会照常执行适合把数据库连接池、后台任务等资源初始化放在注册的 lifespan 任务中。API Transformer 是 Reflex 从纯 Python 前端框架走向全栈后端平台的关键接口它让你无需放弃成熟的 FastAPI 生态即可在同一应用内完成页面渲染与业务 API 的融合。配合 docs/api-routes/overview.md 中的示例与本文的源码解析你可以放心地在生产项目中扩展自己的后端服务。【免费下载链接】reflex️ Web apps in pure Python 项目地址: https://gitcode.com/GitHub_Trending/re/reflex创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考