Python-FastAPI框架使用介绍

见贤思齐

Python-FastAPI框架介绍1.FastAPI简介​FastAPI是一个现代、快速（高性能）的Web框架，用于构建基于Python的API。它具有简单易用的特性，同时也提供了高度自动化的文档生成功能，使得开发者可以更加高效地构建和部署API服务。功能强大、易于使用且高性能的Web框架，适用于构建各种规模的API服务。它的简洁语法、自动生成文档和异步支持等特性使得开发API更加轻松和愉快中文文档：https://fastapi.tiangolo.com/zh/GitHub项目地址：https://github.com/tiangolo/fastapi本篇文章会简单的介绍一下概念，然后通过一个完整的实战项目来展开介绍各个功能模块。FastAPI关键特性快速高效：基于Starlette和Pydantic构建，具有高性能和低延迟的特点，支持异步处理请求，利用Python的协程提高并发性能。简单易用：使用标准的Python类型注解来定义API的输入和输出参数，无需编写大量的文档和验证代码。自动生成文档：通过访问/docs路径可以查看自动生成的交互式API文档，包含了每个端点的详细说明、请求和响应的模型结构以及示例请求和响应。数据验证：利用Pydantic提供的数据验证功能，可以自动验证请求数据的格式和类型，并进行数据转换。依赖注入：支持依赖注入，可以方便地将依赖项注入到处理函数中，例如数据库连接、配置等。类型检查：利用Python类型提示和Pydantic的数据模型，可以在开发过程中进行类型检查，减少错误和调试时间。中间件支持：支持使用中间件扩展框架的功能，例如认证、日志记录等。标准化：支持OpenAPI规范，可以生成符合规范的API文档和客户端代码。2.FastAPI安装安装FastAPI模块pipinstallfastapipipinstalluvicorn12uvicorn用于运行Python的异步Web应用程序，与许多流行的Python框架（如FastAPI、Starlette等）兼容，可以帮助开发者构建高效的异步Web服务3.FastAPI入门体验小试牛刀demo，跑一遍官方小示例创建一个main.py#!/usr/bin/envpython3#-*-coding:utf-8-*-#@Project:wangting_FastAPI#@File:main.py.py#@Author:wangting_666fromtypingimportUnionfromfastapiimportFastAPIapp=FastAPI()@app.get("/")asyncdefread_root():return{"Hello":"World"}@app.get("/items/{item_id}")asyncdefread_item(item_id:int,q:Union[str,None]=None):return{"item_id":item_id,"q":q}1234567891011121314151617181920async关键字用于定义异步函数。异步函数可以在执行过程中暂停并允许其他代码执行，直到某些条件满足后再恢复执行。在FastAPI中，使用async可以使函数能够处理异步操作，例如异步的数据库查询、IO操作等，以提高性能和并发能力。在这个例子中，read_root和read_item函数都是异步函数，它们使用了async关键字来定义。这样的函数可以通过await关键字调用其他异步函数，或者执行需要等待的异步操作，而不会阻塞整个应用程序的执行。12在PyCharm的Terminal中运行uvicorn命令启动FastAPI应用程序uvicornmain:app--reload1这里的main是你的Python文件名（不含扩展名main方法的意思），app是你创建的FastAPI应用程序实例。uvicornmain:app--reload命令含义如下:main：main.py文件（一个Python“模块”）。app：在main.py文件中通过app=FastAPI()创建的对象。--reload：让服务器在更新代码后重新启动。仅在开发时使用该选项。打开浏览器并验证访问：http://localhost:8000/http://localhost:8000/items/1图例1图例2通过以上代码创建了一个具有以下功能的API：通过路径/和/items/{item_id}接受HTTP请求。以上路径都接受GET操作（也被称为HTTP方法）。/items/{item_id}路径有一个路径参数item_id并且应该为int类型。/items/{item_id}路径有一个可选的str类型的查询参数q。4.FastAPI交互式文档访问http://127.0.0.1:8000/docs可打开API文档界面5.FastAPI入门体验代码升级小试牛刀demo，官方小示例迭代main.pyfromtypingimportUnionfromfastapiimportFastAPIfrompydanticimportBaseModelapp=FastAPI()classItem(BaseModel):name:strprice:floatis_offer:Union[bool,None]=None@app.get("/")defread_root():return{"hello":"world"}@app.get("/item/{item_id}")defread_item(item_id:int,q:Union[str,None]=None):return{"item_id":item_id,"q":q}@app.put("/items/{item_id}")defupdate_item(item_id:int,item:Item):return{"item_name":item.name,"item_id":item_id}1234567891011121314151617181920212223242526服务器将会自动重载（因为在上面的步骤中你向uvicorn命令添加了--reload选项）由于之前在PyCharm的Terminal中运行uvicorn命令启动FastAPI应用程序，有--reload参数，所以无需重新运行，如没有增加这个参数，需要在Terminal中重新执行uvicorn打开URLhttp://localhost:8000/docs找到put，点击Tryitout；可以填写参数并直接调用API图例1：点击「Execute」按钮，用户界面将会和API进行通信，发送参数，获取结果并在屏幕上展示图例2：6.FastAPI路径介绍fastapi支持各种请求方式：装饰器HTTP方法说明使用注意事项@app.get()GET处理HTTPGET请求，获取资源的信息或数据。应该是幂等的，不应该对服务器状态进行修改。不要包含对资源的修改操作。@app.post()POST处理HTTPPOST请求，创建新资源或提交数据。POST请求应该包含一个请求体，应避免包含敏感信息。@app.put()PUT处理HTTPPUT请求，更新已存在的资源或创建指定标识的资源。PUT请求应该包含一个完整的资源表示，用于替换原始资源。@app.patch()PATCH处理HTTPPATCH请求，部分更新已存在的资源的内容。PATCH请求应该包含一个用于指定需要更新的部分资源内容的请求体。应避免更新资源的标识或其他关键信息。@app.delete()DELETE处理HTTPDELETE请求，删除指定资源。DELETE请求应该谨慎使用，应使用权限验证和确认机制。@app.options()OPTIONS处理HTTPOPTIONS请求，获取目标资源支持的通信选项。OPTIONS请求通常由浏览器在跨域请求时发送，一般不需要直接处理。@app.head()HEAD处理HTTPHEAD请求，仅返回响应头信息，不返回实际内容。获取资源的响应头信息，而不获取实际的资源内容。HEAD请求与GET请求类似，但不返回实际的资源内容。@app.trace()TRACE处理HTTPTRACE请求，追踪请求在传输链路上的路径，用于调试。TRACE请求通常用于调试和诊断网络问题，一般不在业务逻辑中直接使用。7.项目实战示例1(CRUD)​让我们一起探索一个完整的FastAPI项目实战！在这个项目中，帮助大家深入了解FastAPI框架的强大功能，并通过实际示例演示其灵活性和易用性。项目示例可以提供一个直观、易理解的教程，帮助您快速掌握FastAPI的核心概念和最佳实践。跟随我们一起探索，开启您的FastAPI之旅。通过一个魔兽世界游戏的职业信息库的增删改查需求来慢慢了解各个功能。7-1.项目目录结构app0415app(package包，功能模块)__init__.pylogger.pymain.pytodo.pybin（项目执行入口）run_server.pyconfig（配置文件）config.pylog（日志输出）app.logtmp（存放临时文件用）requirements.txt（模块依赖包清单）7-2.准备工作需要一个MySQL数据库--建测试库CREATEDATABASE`wow`;usewow;#创建测试表CREATETABLE`wow_info`(`id`int(11)NOTNULLAUTO_INCREMENTCOMMENT'角色id',`role`varchar(255)DEFAULTNULLCOMMENT'角色简称',`role_cn`varchar(255)DEFAULTNULLCOMMENT'角色类型',`role_pinyin`varchar(255)DEFAULTNULLCOMMENT'角色拼音',`zhuangbei`varchar(255)DEFAULTNULLCOMMENT'装备类型',`tianfu`varchar(255)DEFAULTNULLCOMMENT'天赋类型',PRIMARYKEY(`id`))ENGINE=InnoDBAUTO_INCREMENT=667DEFAULTCHARSET=utf8;#插入样例数据INSERTINTO`wow_info`VALUES(1,'fs','法师','fashi','布甲','冰法|火法|奥法');INSERTINTO`wow_info`VALUES(2,'ms','牧师','mushi','布甲','神牧|戒律|暗牧');INSERTINTO`wow_info`VALUES(3,'ss','术士','shushi','布甲','毁灭|痛苦|恶魔');INSERTINTO`wow_info`VALUES(4,'dz','盗贼','daozei','皮甲','狂徒|刺杀|敏锐');INSERTINTO`wow_info`VALUES(5,'ws','武僧','wuseng','皮甲','酒仙|踏风|织雾');INSERTINTO`wow_info`VALUES(6,'xd','德鲁伊','xiaode','皮甲','恢复|平衡|野性|守护');INSERTINTO`wow_info`VALUES(7,'dh','恶魔猎手','emolieshou','皮甲','复仇|浩劫');INSERTINTO`wow_info`VALUES(8,'lr','猎人','lieren','锁甲','兽王|生存|射击');INSERTINTO`wow_info`VALUES(9,'sm','萨满','saman','锁甲','恢复|增强|元素');INSERTINTO`wow_info`VALUES(10,'long','龙人','longren','锁甲','湮灭|恩护|增辉');INSERTINTO`wow_info`VALUES(11,'dk','死亡骑士','siwangqishi','板甲','鲜血|冰霜|邪恶');INSERTINTO`wow_info`VALUES(12,'zs','战士','zhanshi','板甲','武器|狂暴|防护');INSERTINTO`wow_info`VALUES(13,'sq','圣骑士','shengqi','板甲','神圣|防护|惩戒');123456789101112131415161718192021222324252627282930Postman有条件建议安装一个Postman接口调试工具，非常实用方便pycharm开发工具和python环境7-3.代码介绍run_server.py#!/usr/bin/envpython3#-*-coding:utf-8-*-#@Project:app0415#@File:run_server.py#@Time:2024/4/1519:46#@Author:wangting_666importuvicornif__name__=="__main__":uvicorn.run("app.main:app",host="127.0.0.1",port=8000,reload=True)1234567891011todo.py#!/usr/bin/envpython3#-*-coding:utf-8-*-#@Project:app0415#@File:todo.py#@Time:2024/4/1519:46#@Author:wangting_666fromtypingimportList,OptionalfrompydanticimportBaseModelimportpymysql#MySQL连接配置config={'host':'wangting_host_ip','port':3306,'user':'root','password':'123456','database':'wow',}#连接到MySQL数据库defconnect_to_mysql():returnpymysql.connect(**config)#定义WowInfo模型classWowInfo(BaseModel):id:introle:strrole_cn:strrole_pinyin:strzhuangbei:strtianfu:str#获取所有魔兽职业信息defget_wowinfo_all()->List[WowInfo]:try:conn=connect_to_mysql()withconn.cursor(pymysql.cursors.DictCursor)ascursor:cursor.execute("SELECT*FROMwow_info")info=[WowInfo(**row)forrowincursor.fetchall()]returninfoexceptExceptionase:print(f"查不到职业信息:{e}")return[]#获取单个魔兽职业信息defget_wowinfo(role:str)->Optional[WowInfo]:try:conn=connect_to_mysql()withconn.cursor(pymysql.cursors.DictCursor)ascursor:cursor.execute("SELECT*FROMwow_infoWHERErole=%s",(role,))info=cursor.fetchone()returnWowInfo(**info)ifinfoelseNoneexceptExceptionase:print(f"查不到职业信息:{e}")returnNone#创建魔兽职业信息defcreate_wowinfo(wowinfo:WowInfo)->WowInfo:try:conn=connect_to_mysql()withconn.cursor()ascursor:cursor.execute("INSERTINTOwow_info(id,role,role_cn,role_pinyin,zhuangbei,tianfu)VALUES(%s,%s,%s,%s,%s,%s)",(wowinfo.id,wowinfo.role,wowinfo.role_cn,wowinfo.role_pinyin,wowinfo.zhuangbei,wowinfo.tianfu))conn.commit()returnwowinfoexceptExceptionase:print(f"创建职业信息失败:{e}")returnNone#更新魔兽职业信息defupdate_wowinfo(id:int,wowinfo:WowInfo)->Optional[WowInfo]:try:conn=connect_to_mysql()withconn.cursor()ascursor:cursor.execute("UPDATEwow_infoSETrole=%s,role_cn=%s,role_pinyin=%s,zhuangbei=%s,tianfu=%sWHEREid=%s",(wowinfo.role,wowinfo.role_cn,wowinfo.role_pinyin,wowinfo.zhuangbei,wowinfo.tianfu,id))conn.commit()returnwowinfoexceptExceptionase:print(f"更新职业信息失败:{e}")returnNone#删除魔兽职业信息defdelete_wowinfo(id:int)->bool:try:conn=connect_to_mysql()withconn.cursor()ascursor:cursor.execute("DELETEFROMwow_infoWHEREid=%s",(id,))conn.commit()returnTrueexceptExceptionase:print(f"删除职业信息失败:{e}")returnFalse123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103main.py#!/usr/bin/envpython3#-*-coding:utf-8-*-#@Project:app0415#@File:main.py#@Time:2024/4/1519:46#@Author:wangting_666fromfastapiimportFastAPI,HTTPExceptionfromconfig.configimportsettingsfromapp.loggerimportloggerfromapp.todoimportWowInfo,get_wowinfo_all,get_wowinfo,create_wowinfo,update_wowinfo,delete_wowinfofromtypingimportListapp=FastAPI(title=settings.app_name)@app.get("/get_wowinfo_all/",response_model=List[WowInfo])asyncdefread_wowinfos():logger.info("查询所有魔兽世界角色...")returnget_wowinfo_all()@app.get("/get_wowinfo/{role}",response_model=WowInfo)asyncdefread_wowinfo(role:str):logger.info(f"查询魔兽世界角色名称为:{role}...")info=get_wowinfo(role)ifinfoisNone:logger.error(f"角色{role}不存在.")raiseHTTPException(status_code=404,detail="查询失败")returninfo@app.post("/create_wowinfo/",response_model=WowInfo)asyncdefcreate_new_wowinfo(info:WowInfo):logger.info("创建魔兽世界角色")returncreate_wowinfo(info)@app.put("/update_wowinfo/{id}",response_model=WowInfo)asyncdefupdate_existing_wowinfo(id:int,wowinfo:WowInfo):logger.info(f"更新魔兽世界角色id{id}...")existing_info=update_wowinfo(id,wowinfo)ifexisting_infoisNone:logger.error(f"职业信息ID:{id}不存在")raiseHTTPException(status_code=404,detail="职业信息ID不存在")returnexisting_info@app.delete("/delete_wowinfo/{id}")asyncdefdelete_existing_wowinfo(id:int):logger.info(f"删除魔兽世界角色id{id}...")success=delete_wowinfo(id)ifnotsuccess:logger.error(f"兽世界角色id{id}不存在")raiseHTTPException(status_code=404,detail="id不存在")return{"status":"success","message":"删除魔兽世界角色id{id}成功"}1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556logger.py#!/usr/bin/envpython3#-*-coding:utf-8-*-#@Project:app0415#@File:logger.py#@Time:2024/4/1519:46#@Author:wangting_666importloggingfromconfig.configimportsettingsimportoslog_dir=os.path.join(os.path.dirname(os.path.abspath(__file__)),"../log")os.makedirs(log_dir,exist_ok=True)logging.basicConfig(filename=os.path.join(log_dir,"app.log"),level=settings.log_level)logger=logging.getLogger(__name__)12345678910111213141516config.py#!/usr/bin/envpython3#-*-coding:utf-8-*-#@Project:app0415#@File:config.py#@Time:2024/4/1519:46#@Author:wangting_666frompydanticimportBaseSettingsclassSettings(BaseSettings):app_name:str="TodoApp"log_level:str="INFO"settings=Settings()123456789101112131415167-4.运行服务进行验证7-4-1.启动服务pycharm方式我这里是通过pycharm方式，如果是服务器运行，只需要执行python3/home/wangting/run_server.py运行起来即可开始功能测试之前，可以通过上方介绍的/docs接口查看自动生成的接口文档http://127.0.0.1:8000/docs7-4-2.验证服务后续验证服务，get请求可以通过浏览器、postman、接口的/docs功能均可；我这里使用postman进行演示，也可以通过上方介绍的/docs中tryitout中进行execute进行调试验证get_wowinfo_allget_wowinfo_all对应了get_wowinfo_all方法，@app.get可以看出是get请求，路由是/get_wowinfo_all@app.get("/get_wowinfo_all/",response_model=List[WowInfo])asyncdefread_wowinfos():logger.info("查询所有魔兽世界角色...")returnget_wowinfo_all()1234/get_wowinfo/{role}/get_wowinfo/{role}对应了get_wowinfo方法，@app.get可以看出也是get请求，路由是/get_wowinfo_all/{role}@app.get("/get_wowinfo/{role}",response_model=WowInfo)asyncdefread_wowinfo(role:str):logger.info(f"查询魔兽世界角色名称为:{role}...")info=get_wowinfo(role)ifinfoisNone:logger.error(f"角色{role}不存在.")raiseHTTPException(status_code=404,detail="查询失败")returninfo12345678create_wowinfocreate_wowinfo对应了create_wowinfo方法，@app.post可以看出是post请求，路由是/create_wowinfo/但与get请求不同的是，post请求需要提供body请求体{"id":666,"role":"create_role","role_cn":"create_role_cn","role_pinyin":"create_role_pinyin","zhuangbei":"create_zhuangbei","tianfu":"create_tianfu"}12345678@app.post("/create_wowinfo/",response_model=WowInfo)asyncdefcreate_new_wowinfo(info:WowInfo):logger.info("创建魔兽世界角色")returncreate_wowinfo(info)1234此时查看一下MySQL数据库，验证是否真的写入到数据库update_wowinfo/{id}update_wowinfo/{id}对应了update_wowinfo方法，@app.put可以看出是put请求，路由是/update_wowinfo/{id}与post请求一样，也需要提供body请求体{"id":666,"role":"update_role","role_cn":"update_role_cn","role_pinyin":"update_role_pinyin","zhuangbei":"update_zhuangbei","tianfu":"update_tianfu"}12345678@app.put("/update_wowinfo/{id}",response_model=WowInfo)asyncdefupdate_existing_wowinfo(id:int,wowinfo:WowInfo):logger.info(f"更新魔兽世界角色id{id}...")existing_info=update_wowinfo(id,wowinfo)ifexisting_infoisNone:logger.error(f"职业信息ID:{id}不存在")raiseHTTPException(status_code=404,detail="职业信息ID不存在")returnexisting_info12345678查询数据库数据，id=666的数据内容已经被修改MariaDB[wow]>select*fromwow_infowhereid=666\G;***************************1.row***************************id:666role:update_rolerole_cn:update_role_cnrole_pinyin:update_role_pinyinzhuangbei:update_zhuangbeitianfu:update_tianfu12345678delete_wowinfo/{id}delete_wowinfo/{id}对应了delete_wowinfo方法，@app.delete可以看出是delete请求，路由是/delete_wowinfo/{id}@app.delete("/delete_wowinfo/{id}")asyncdefdelete_existing_wowinfo(id:int):logger.info(f"删除魔兽世界角色id{id}...")success=delete_wowinfo(id)ifnotsuccess:logger.error(f"兽世界角色id{id}不存在")raiseHTTPException(status_code=404,detail="id不存在")return{"status":"success","message":"删除魔兽世界角色id{id}成功"}12345678此时查询数据库数据id=666的数据内容已经不存在了MariaDB[wow]>select*fromwow_infowhereid=666\G;Emptyset(0.000sec)127-4-3.项目实战总结​这个FastAPI项目展示了一个完整的后端应用程序，用于管理魔兽世界角色的信息。通过这个项目，我们深入了解了FastAPI框架的使用，以及如何构建RESTfulAPI。​首先，我们学习了如何定义数据模型，使用Pydantic创建了WowInfo模型来表示魔兽世界角色的信息。然后，我们编写了一系列处理CRUD操作的函数，用于获取、创建、更新和删除角色信息。这些函数与MySQL数据库进行交互，通过pymysql库执行SQL查询和操作。​接下来，我们使用FastAPI创建了一个基于路由的Web服务，定义了不同的端点来处理不同的HTTP请求。通过最常用的装饰器@app.get、@app.post、@app.put和@app.delete，我们将每个端点与对应的处理函数绑定在一起。​最后，我们使用了日志记录功能和配置设置，使项目更具可维护性和可扩展性。我们将日志记录到文件中，并使用配置类来管理应用程序的设置。​这个项目为我们提供了一个完整的FastAPI实战示例，展示了如何使用FastAPI框架构建一个简单但功能完备的后端应用程序。通过这个示例，我们不仅学习了FastAPI框架的核心概念和基本用法，还掌握了与数据库交互、路由定义、日志记录等高级功能。8.项目实战示例2(前后端)8-1.项目目录结构app0416server_monitor_frontendindex.htmlserver_monitor_backend.py示例是一个前后端协作，仅展示逻辑demo，没有加入太多的功能模块在app0416目录下有一个名为server_monitor_backend.py的文件，用于启动后端FastAPI服务器。在同一目录下还有一个名为server_monitor_frontend的文件夹，其中包含一个名为index.html的文件，用于显示服务器监控系统的界面。如果需要自定义样式，可以在server_monitor_frontend文件夹中创建一个名为style.css的文件，并在index.html中引入，此目录主要用来开发前端功能。最终实现打开http://127.0.0.1:8000/，可以5秒刷新展示一次系统的服务器资源情况。8-2.代码介绍server_monitor_backend.py#!/usr/bin/envpython3#-*-coding:utf-8-*-#@Project:app0416#@File:server_monitor_backend.py.py#@Time:2024/4/1619:58#@Author:wangting_666fromfastapiimportFastAPIfromfastapi.staticfilesimportStaticFilesfromstarlette.responsesimportHTMLResponse,FileResponseapp=FastAPI()#将静态文件目录设置为server_monitor_frontend目录app.mount("/static",StaticFiles(directory="server_monitor_frontend"),name="static")@app.get("/",response_class=HTMLResponse)asyncdefget_index_page():returnFileResponse("server_monitor_frontend/index.html")@app.get("/stats")asyncdefget_server_stats():#Yourserverstatslogicherereturn{"cpu_percent":50,"memory_percent":60,"disk_percent":70}if__name__=="__main__":importuvicornuvicorn.run(app,host="127.0.0.1",port=8000)1234567891011121314151617181920212223242526272829303132index.html ServerMonitor 123456789101112131415161718192021222324252627282930313233348-3.运行服务进行验证运行server_monitor_backend.py打开URL：http://127.0.0.1:8000/8-4.前后端示例总结这个案例演示了如何使用FastAPI构建一个简单的服务器监控系统。后端FastAPI提供了/stats路由，用于获取服务器的监控数据，而前端界面通过JavaScript定时获取数据并动态显示在页面上。这个示例结合了后端和前端技术，展示了如何利用FastAPI快速构建API，以及如何使用JavaScript实现动态页面交互。