FastAPI 是一个用于构建 API 的现代、快速(高性能)的 web 框架。 入门最小程序下面代码会直接启动http服务,也可以使用 uvicorn main:app --reload from fastapi import FastAPI import uvicorn
app = FastAPI()
# http://127.0.0.1:8000/ @app.get('/') async def root(): return {'message': 'Hello World'}
if __name__ == '__main__': uvicorn.run(app='main:app', reload=True)
路径参数最基本的路径参数# http://127.0.0.1:8000/items/1 @app.get('/items/{item_id}') async def read_item(item_id): return {'item_id': item_id} # item_id自定义
多个路径参数# http://127.0.0.1:8000/items/1/2 @app.get('/items/{item_id}/{user_id}') async def read_item(item_id, user_id): return {'item_id': item_id, 'user_id': user_id}
有类型的路径参数# http://127.0.0.1:8000/items/1 @app.get('/items/{item_id}') async def read_item(item_id: int): return {'item_id': item_id}
文件路径参数# http://127.0.0.1:8000/file//home/my/my.txt @app.get('/file/{file_path:path}') async def read_item(file_path): return {'file_path': file_path}
查询参数带默认值的查询参数# http://127.0.0.1:8000/items/?skip=0&limit=2 fake_items_db = [{'item_name': 'Foo'}, {'item_name': 'Bar'}, {'item_name': 'Baz'}] @app.get('/items/') async def read_item(skip: int = 0, limit: int = 10): return fake_items_db[skip: skip + limit]
可选查询参数# http://127.0.0.1:8000/items/1?q=admin from typing import Union @app.get('/items/{item_id}') async def read_item(item_id: str, q: Union[str, None] = None): if q: return {'item_id': item_id, 'q': q} return {'item_id': item_id}
多路径多查询参数# http://127.0.0.1:8000/users/1/items/2 # or # http://127.0.0.1:8000/users/1/items/2?q=query&short=true @app.get('/users/{user_id}/items/{item_id}') async def read_user_item( user_id: int, item_id: str, q: Union[str, None] = None, short: bool = False ): item = {'item_id': item_id, 'owner_id': user_id} if q: item.update({'q': q}) if not short: item.update( {'description': 'This is an amazing item that has a long description'} ) return item
必需查询参数# http://127.0.0.1:8000/items/123?needy=yes @app.get('/items/{item_id}') async def read_user_item(item_id: str, needy: str): item = {'item_id': item_id, 'needy': needy} return item
请求体from pydantic import BaseModel from typing import Union class Item(BaseModel): name: str = '小明' description: Union[str, None] = None price: float tax: Union[float, None] = None @app.post('/items/') async def create_item(item: Item): print(item.name) return item
调用curl -X 'POST' \ 'http://127.0.0.1:8000/items/' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{ 'name': '小明', 'description': 'string', 'price': 0, 'tax': 0 }'
查询参数和字符串校验from fastapi import Query @app.get('/items/') async def read_items(q: Union[str, None] = Query(default=None, max_length=50)): results = {'items': [{'item_id': 'Foo'}, {'item_id': 'Bar'}]} if q: results.update({'q': q}) return results
参数列表参数 | 含义 | 类型 | default | 默认值 | 任意类型或... | max_length | 最大长度 | int | min_length | 最小长度 | int | pattern | 正则匹配 | string | alias | 别名参数 | string | deprecated | 准备弃用参数 | bool |
多个相同的查询参数# http://127.0.0.1:8000/items/?q=foo&q=bar @app.get('/items/') async def read_items(q: Union[List[str], None] = Query(default=None)): query_items = {'q': q} return query_items
路径参数和数值校验Path用法基本和Query相同 导入 Pathfrom fastapi import FastAPI, Path, Query from typing_extensions import Annotated
声明元数据@app.get('/items/{item_id}') async def read_items( item_id: Annotated[int, Path(title='The ID of the item to get')], q: Annotated[Union[str, None], Query(alias='item-query')] = None, ): results = {'item_id': item_id} if q: results.update({'q': q}) return results
参数列表参数 | 含义 | 类型 | ... | 和Query具有相同参数 | ... | ge | 大于等于 | int float | gt | 大于 | int float | le | 小于等于 | int float | le | 小于等于 | int float | title | api文档的标题 | string |
其他参数都具有Query的参数,max_length、min_length等 Cookie参数from fastapi import Cookie @app.get('/items/') async def read_items(ads_id: Annotated[Union[str, None], Cookie()] = None): return {'ads_id': ads_id}
Header 参数from fastapi import Header @app.get('/items/') async def read_items(user_agent: Annotated[Union[str, None], Header()] = None,items_id: Annotated[Union[int, None], Header(ge=1)] = None): return {'User-Agent': user_agent, 'items_id': items_id}
表单数据接收的不是 JSON,而是表单字段时,要使用 Form。 安装pip install python-multipart
HTML<!DOCTYPE html> <html lang='en'> <head> <meta charset='UTF-8'> </head> <body> <form method='post' action='http://127.0.0.1:8000/login'> <span>账号:</span><input type='text' name='username'> <br> <span>密码:</span><input type='password' name='password'> <br> <input type='submit' value='登录'> </form> </body> </html>
FastAPIfrom fastapi import FastAPI, Form import uvicorn app = FastAPI() @app.post('/login/') async def login(username: str = Form(), password: str = Form()): return {'username': username} if __name__ == '__main__': uvicorn.run(app='main:app', reload=True)
文件上传from fastapi import FastAPI, UploadFile from fastapi.responses import HTMLResponse @app.post('/uploadfile/') async def create_upload_file(file: UploadFile): print(file.file.read().decode()) return {'filenames': file.filename, 'type': str(type(file.file))} @app.get('/') async def main(): content = '''<body> <form action='/uploadfile/' enctype='multipart/form-data' method='post'> <input name='file' type='file' multiple> <input type='submit'> </form> </body>''' return HTMLResponse(content=content)
UploadFile属性属性名 | 含义 | 返回 | filename | 文件名 | 上传的文件名 | content_type | 内容类型 | MIME 类型 | file | 文件 | SpooledTemporaryFile具有read,write方法 |
UploadFile async方法方法名 | 含义 | write(data) | 把 data 写入文件 | read(size) | 按指定数量的字节读取文件内容 | seek(offset) | 移动至文件 offset (int )字节处的位置 | close() | 关闭文件 |
依赖项依赖项使用场景· 共享业务逻辑(复用相同的代码逻辑) · 共享数据库连接 · 实现安全、验证、角色权限 · 等……
创建依赖项from typing import Union
from fastapi import Depends, FastAPI
app = FastAPI()
# read_items和read_users方法依赖common_parameters # 白话就是read_items和read_users都需要q,skip,limit查询参数 async def common_parameters( q: Union[str, None] = None, skip: int = 0, limit: int = 100 ): return {'q': q, 'skip': skip, 'limit': limit}
@app.get('/items/') async def read_items(commons: dict = Depends(common_parameters)): return commons
@app.get('/users/') async def read_users(commons: dict = Depends(common_parameters)): return commons
类作为依赖项from typing import Union
from fastapi import Depends, FastAPI
app = FastAPI()
fake_items_db = [{'item_name': 'Foo'}, {'item_name': 'Bar'}, {'item_name': 'Baz'}]
class CommonQueryParams: def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100): self.q = q self.skip = skip self.limit = limit
# read_itemsx接收一个commons参数,类型是CommonQueryParams # CommonQueryParams接收三个参数,这三个参数是调用api的时候传 @app.get('/items/') async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)): response = {} if commons.q: response.update({'q': commons.q}) items = fake_items_db[commons.skip : commons.skip + commons.limit] response.update({'items': items}) return response
还可以简写@app.get('/items/') async def read_items(commons: CommonQueryParams = Depends()): # 这里的Depends没有传参,FastAPI会自动使用CommonQueryParams response = {} if commons.q: response.update({'q': commons.q}) items = fake_items_db[commons.skip : commons.skip + commons.limit] response.update({'items': items}) return response
子依赖项from typing import Union
from fastapi import Cookie, Depends, FastAPI
app = FastAPI()
def query_extractor(q: Union[str, None] = None): return q
def query_or_cookie_extractor( q: str = Depends(query_extractor), last_query: Union[str, None] = Cookie(default=None), ): if not q: return last_query return q
# read_query函数依赖query_or_cookie_extractor函数 # query_or_cookie_extractor函数又依赖query_extractor函数 # 就是说依赖项可以依赖其他依赖项,只要你不晕,可以无数次套娃 @app.get('/items/') async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)): return {'q_or_cookie': query_or_default}
不使用缓存# 使用use_cache = False参数不使用缓存数据 # 不使用use_cache = False,value和value1是一样的 def result_value(): value = randint(1, 99) return value
def get_value(value: int = Depends(result_value, use_cache=False), value1: int = Depends(result_value, use_cache=False)): return value, value1
@app.get('/value/') async def needy_dependency(value: tuple = Depends(get_value)): return {'value': value}
全局依赖项from fastapi import Depends, FastAPI, Header, HTTPException
async def verify_token(x_token: str = Header()): if x_token != 'fake-super-secret-token': raise HTTPException(status_code=400, detail='X-Token header invalid')
async def verify_key(x_key: str = Header()): if x_key != 'fake-super-secret-key': raise HTTPException(status_code=400, detail='X-Key header invalid') return x_key
# 全局依赖项很有用,后面的安全性就可以使用全局依赖项 app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])
@app.get('/items/') async def read_items(): return [{'item': 'Portal Gun'}, {'item': 'Plumbus'}]
@app.get('/users/') async def read_users(): return [{'username': 'Rick'}, {'username': 'Morty'}]
|