FastAPI-路由机制和依赖注入

FastAPI-路由机制和依赖注入1.绪言今天写项目被FastAPI的这个路由机制给干懵逼了啥叫router给app加上一个include_router是干啥用的装饰器又是啥严肃意识到自己的基础仍然非常薄弱于是整理下面的笔记以供后来参考2.Why路由在一开始对FastAPI的使用中我们一般都会写出这样的代码fromfastapiimportFastAPI appFastAPI()app.get(/)defget_item():return{status:ok}是吧虽然一开始我也不知道这是什么原理但只需要知道上面app的方法里填路径下面写一个处理函数这个函数就会被注册到对应的路径下访问时就会执行对应逻辑但是我们知道所有的架构都是需要复杂的项目逻辑来验证的当逻辑变得更加复杂之后defhandle_request(request):ifrequest.pathusers/andrequest.methodGET:returnget_all_users()elifrequest.pathitems/andrequest.methodPOST:returncreate_item()# ...若干我们知道如果代码里写了非常多个if-else那么基本可以把它归为垃圾一类了所以总之如果单纯采用手动判断转发的方式无疑会极大程度上影响结构和逻辑清晰度此外路由机制可以将不同逻辑解耦到不同文件中而不至于全部塞到main.py里堆成一座屎山3.理解app是什么3.1.概念阐述在一开始我提到FastAPI的app只能模糊说出这大概是一个总体的应用程序什么鬼话事实上app可以理解为整个后端应用的总路由表 请求处理入口当你的浏览器或者说测试代码请求一个GET /healthFastAPI会问我的app里有没有注册过一个路径是/health方法是GET的处理函数如果有就调用那个函数app FastAPI()创建了一个ASGI应用对象然后uvicorn监听端口接受HTTP请求转发给app然后app做路由匹配、参数解析、调用函数、生成响应3.2.关于网络端口写到这里发现网络端口这块也需要讲一下当主机连接到互联网中时会被分配一个IP地址例如像192.168.x.x这样的代表在这个网络内这个地址就指向你的主机而在这个IP地址后面有时我们会看到类似于192.168.x.x:8000这样的东西这是什么后面的那个8000就叫做端口在你的主机上运行着很多应用程序然后其中很多应用程序都需要接入网络然后我们会想到肯定不能把它们全部接在一起这样一来不好管理所以我们引入了端口的概念每个应用程序或者说进程被分配到一个端口用来和外界网络连接我们可以想象就像进程在堆区开辟属于它自己的一块内存资源空间一样每个程序也相当于独占了一个房间FastAPI在这个房间跑着服务器React在那个房间渲染网页像微信QQ这些也都占据着各自的房间此时为了区分到底哪个房间是FastAPI在的地方端口就像门牌号标在上面外面一看就知道应该把哪些请求哪些资源送到哪里总而言之IP决定数据发往哪栋楼而端口决定数据发往哪个房间这里不再详述了感觉以后可以专门写一篇等我计网什么的通了从一个主机到服务器再到主机这个过程每一步都发生了什么立个flag4.路由装饰器4.1.装饰器4.1.1.装饰器基本介绍在讲路由装饰器之前我们先需要阐明一下Python装饰器的定义对于Python的装饰器它实现的功能其实很简单就是给一个已有的函数添加功能假如说我现在需要测试某个函数运行花费了多少时间importtimedefwaste_time(n:int):foriinrange(n):print(i)defcheck_cost_time(func):start_timetime.time()func()end_timetime.time()costend_time-start_timeprint(f花费{cost}秒)这样写虽然也可以实现测量运行时间的功能但问题在于它并不是给原有函数添加功能在意图和语义上和我们所期望的有所偏差所以我们可以使用装饰器importtimedeftimer(func):defwrapper(*args,**kwargs):start_timetime.time()func(*args,**kwargs)end_timetime.time()costend_time-start_timeprint(f花费{cost}秒)returnwrappertimerdefwaste_time(n:int):foriinrange(n):print(i)# 然后直接调用waste_time(114514)这里的其实就只是一个语法糖它实际上实现了waste_time timer(waste_time)4.1.2.*args和**kwargs我们注意到在前面的wrapper包裹函数中参数列表里提到了*args和**kwargs这两个其实是Python提供的变长参数可以自动匹配任意多个参数*args用于接收位置参数(Positional Arguments)在函数内部打包成一个元组例如defget_sum(a,b,c0,d0):returnsum((a,b,c,d))# 不可能预知有多少个参数defget_sum(*args):returnsum(args)print(get_sum(1,2,3,4,5))# 传多少个都可以**kwargs用于接收关键字参数(Keywords Arguments)在函数内部打包成一个字典例如defconfigure_agent(**kwargs):print(fkwargs的类型是{type(kwargs)}内容是{kwargs})ifllm_modelinkwargs:print(f正在启动模型{kwargs[llm_model]})configure_agent(llm_modelgemini,apixxx,token1919810)这里其实不一定要叫args和kwargs只是一种约定俗成的命名规范你用*wsvsbyellowds什么的也是可以的但就像self你非要写成mine一样很容易被打额外讲一点其实*和**不止能用来打包还能用来解包例如defconnect_db(host,port,user):print(f成功连接到{host}:{port}用户{user})db_tuple(127.0.0.1,8000,Abel)# 元组db_dict{host:127.0.0.1,port:8000,user:Abel}# 字典# 解包connect_db(*db_tuple)connect_db(**db_dict)4.1.3.带参数的装饰器装饰器本身也可以接收参数比如说指定日志的级别deflogger(level):defdecorator(func):defwrapper(*args,**kwargs):print(f{level.upper()}开始执行{func.__name__})returnfunc(*args,**kwargs)returnwrapperreturndecoratorlogger(levelinfo)defapi_call(api):print(API正在调用...)api_call(xxx)4.1.4.元数据丢失需要注意的是当使用装饰器装饰一个函数时这个函数的元数据会丢失例如__name__,__doc__timerdeffoo():pass此时调用foo.__name__拿到的不会是foo而是wrapper为了解决这个问题我们可以应用functools.wraps将数据拷贝回来fromfunctoolsimportwrapsdefbetter_timer(func):wraps(func)defwrapper(*args,**kwargs):returnfunc(*args,**kwargs)returnwrapper4.1.5.类装饰器除了用函数实现装饰器以外还可以用类来实现只需要这个类实现了__call__方法classCounter:def__init__(self,func):self.funcfunc self.count0def__call__(self,*args,**kwargs):self.count1print(f已经被调用了{self.count}次)returnself.func(*args,**kwargs)Counterdefsubmit():pass装饰器是个非常有意思的东西但在这里我们就不扩展讲太多了4.2.路由装饰器4.2.1.从示例开始现在我们回到路由装饰器本身举个例子app.get(/health)asyncdefhealth_check():return{status:ok}在这里app.get(/health)做了三件事记录HTTP方法GET记录URL路径/health记录事件处理函数health_check因此FastAPI内部会记录这样一条规则GET /health - health_check()当请求进来时触发GET /health调用awaithealth_check()4.2.2.HTTP请求与RESTful规范4.2.2.1.HTTP请求关于HTTP请求我们这里暂时只讲HTTP请求方法像上文app.get()里的get就是HTTP请求的一种HTTP方法FastAPI中对应的装饰器语义GETapp.get()获取/读取资源POSTapp.post()创建/提交资源PUTapp.put()完整更新/替换资源DELETEapp.delete()删除资源PATCHapp.patch()局部更新资源修补4.2.2.2.RESTful规范它实际上是一种API架构设计规范风格核心思想就是把网络上的所有东西都看作“资源”Resource用“统一的接口”去操作它们有四大准则用“名词”表示资源(URL设计)用 HTTP 方法Method表示“动作”善用 URL 路径表达层次关系状态码(Status Codes)合理设计4.2.3.路径参数所以我们现在就可以理解了路由装饰器中你使用了什么函数比如app.post()就对应着HTTP请求方法(POST)且在像get()这样的括号里能填的不止有静态的URL字符串路径变量将路由动态化用{}占位同时要在函数参数里声明同名变量底层Pydantic实现校验app.get(/users/{user_id})defget_user(user_id:int):return{user_id:user_id,status:Abel Coding}响应状态码反应操作成功后的状态码app.post(/user,status_code201)defcreate_user():return{status:创建成功}标签与文档FastAPI会自动生成Swagger文档在路由装饰器里写的参数可以用于标识处理函数的功能等app.post(/run-query,tags[Agent核心接口],summary运行Agent的查询,description这是用来触发AI Agent的接口)defrun_query():return{result:success}5.解耦——路由分发我们前面也有提到如果把所有接口都写在main.py很快会变成这样app.get(/health)asyncdefhealth_check():...app.get(/api/problems)asyncdeflist_problems():...app.post(/api/problems)asyncdefcreate_problem():...app.get(/api/contests)asyncdeflist_contests():...app.post(/api/agents)asyncdefcreate_agent():...我们很容易想到应当将不同的业务逻辑拆分到多个不同的文件中那怎么实现app这个应用入口和其他业务逻辑联系在一起呢FastAPI提供了APIRouter假如说以我现在在写的Multi-Agent-Algorithmic-Arena项目为例/problems和/contests显然是两个不同的路由我们现在来给它做一个拆分routers/problems.pyfromfastapiimportAPIRouter,Requestfromapp.databaseimportget_dbfromsqlalchemyimportselectfromapp.models.problemimportProblem routerAPIRouter(prefix/api/problems,tags[problems])router.get(/{problem_id})asyncdefget_problem(problem_id:int):# 这里先省略session会话的获取细节后面会讲session??? statementselect(Problem).where(Problem.idproblem_id)resultawaitsession.execute(statement)returnresultrouter.post(/)asyncdefpost_problem(request:Request):problemhandle_request(request)# 假设有这样一个函数session??? session.add(problem)awaitsession.commit()routers/contests.pyfromfastapiimportAPIRouter,Requestfromapp.databaseimportget_dbfromsqlalchemyimportselectfromapp.models.contestimportContest routerAPIRouter(prefix/api/contests,tags[contests])router.get(/{contest_id})asyncdefget_contest(contest_id:int):session??? statementselect(Contest).where(Contest.idcontest_id)resultawaitsession.execute(statement)returnresultrouter.post(/)asyncdefpost_contest(request:Request):contesthandle_request(request)session??? session.add(contest)awaitsession.commit()main.pyfromapp.routers.problemsimportrouterasproblem_router# 引入路由fromapp.routers.contestsimportrouterascontest_routerfromfastapiimportFastAPI appFastAPI()app.include_router(problem_router)# 将路由挂到app上app.include_router(contest_router)app.get(/health)asyncdefcheck_health():return{status:ok}通过这样的方式我们就成功将业务逻辑解耦分发到不同模块的路由上了对于problems和contests各自的router可以理解为它们各自模块的一个小路由表或者说小app6.Depends6.1.从需求出发回顾我们上面的代码在GET方法中我们需要从数据库中查找题目所以想要一个数据库session会话asyncdefget_problem():session???现在问题在于这个session应当从哪里来由谁创建由谁关闭每个请求是不是需要一个新的session出错怎么管理对于这些逻辑如果我们在每个路由中都这样手动去写sessionasync_session()try:...finally:awaitsession.close()会重复也可能容易漏不方便统一管理所以FastAPI提供了Depends通过Depends路由函数只需要声明我需要什么然后FastAPI在请求期间准备传入结束后统一管理6.2.最小心智模型我们的get_db()类似于这样asyncdefget_db():asyncwithasync_session()assession:yieldsession这是一个资源申请分配释放的完整过程我们不希望通过复杂而难以管理的手动逻辑去管理它所以通过Depends我们的get_problem就变成asyncdefget_problem(session:AsyncSessionDepends(get_db))它不代表session的默认值是Depends(get_db)而是FastAPI看到这个参数之后并不让客户端传session而是自己去调用get_db()把结果传给session请求流程如下client.get(/api/problems) ↓ FastAPI 匹配 GET /api/problems ↓ 发现路由函数需要 session ↓ 发现 session 来自 Depends(get_db) ↓ 执行 get_db() ↓ yield 出 AsyncSession ↓ 调用 get_problems(session这个 AsyncSession) ↓ 函数执行数据库查询 ↓ 请求结束后回到 get_db()自动退出 async with关闭 session6.3.使用示例我们先从一个玩具例子出发fromfastapiimportDepends,FastAPI appFastAPI()defget_current_user():returnAbelapp.get(/me)defread_user(user:strDepends(get_current_user)):return{user:user}在这里user不来自URL不来自query而是从依赖函数中来所以现在我们修改上面的代码routers/problems.pyfromfastapiimportAPIRouter,Depends,Requestfromapp.databaseimportget_dbfromsqlalchemyimportselectfromapp.models.problemimportProblemfromsqlalchemy.ext.asyncioimportAsyncSession routerAPIRouter(prefix/api/problems,tags[problems])router.get(/{problem_id})asyncdefget_problem(problem_id:int,session:AsyncSessionDepends(get_db)):statementselect(Problem).where(Problem.idproblem_id)resultawaitsession.execute(statement)returnresultrouter.post(/)asyncdefpost_problem(request:Request,session:AsyncSessionDepends(get_db)):problemhandle_request(request)# 假设有这样一个函数session.add(problem)awaitsession.commit()routers/contests.pyfromfastapiimportAPIRouter,Depends,Requestfromapp.databaseimportget_dbfromsqlalchemyimportselectfromapp.models.contestimportContestfromsqlalchemy.ext.asyncioimportAsyncSession routerAPIRouter(prefix/api/contests,tags[contests])router.get(/{contest_id})asyncdefget_contest(contest_id:int,session:AsyncSessionDepends(get_db)):statementselect(Contest).where(Contest.idcontest_id)resultawaitsession.execute(statement)returnresultrouter.post(/)asyncdefpost_contest(request:Request,session:AsyncSessionDepends(get_db)):contesthandle_request(request)session.add(contest)awaitsession.commit()Depends不止能依赖函数还能往里面塞依赖类和其他Depends的链式调用等等非常灵活但这里先不说了因为我也还不会