devbox
菜鸟追梦旅行
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

FastAPI官方教程太棒了（下）

作者：菜鸟追梦旅行 | 2024-04-13 12:41:06

赞

踩

@app.post("/items/", status_code=201) postman

响应状态码

在@app.post()方法中添加status_code参数：

  1. from fastapi import FastAPI
  2.  
  3. app = FastAPI()
  4.  
  5.  
  6. @app.post("/items/", status_code=201)
  7. async def create_item(name: str):
  8.     return {"name": name}

status_code也可以是IntEnum，比如Python的http.HTTPStatus。

常见响应状态码：

  • 100以上，信息；很少直接使用；

  • 200以上，成功；200是OK，201是Created，204是No Content；

  • 300以上，重定向；304是Not Modified；

  • 400以上，客户端错误；404是Not Found；

  • 500以上，服务器错误；

FastAPI引入了status，可以方便的录入这些状态：

  1. from fastapi import FastAPI, status
  2.  
  3. app = FastAPI()
  4.  
  5.  
  6. @app.post("/items/", status_code=status.HTTP_201_CREATED)
  7. async def create_item(name: str):
  8.     return {"name": name}

表单数据

为了使用表单，首先需要安装python-multipart：

pip install python-multipart

示例：

  1. from fastapi import FastAPI, Form
  2.  
  3. app = FastAPI()
  4.  
  5.  
  6. @app.post("/login/")
  7. async def login(username: str = Form(), password: str = Form()):
  8.     return {"username": username}

表单由HTML中的<form></form>发送，请求头的content-type一般是application/x-www-form-urlencoded，当为文件时multipart/form-data。

请求文件

示例：

  1. from fastapi import FastAPI, File, UploadFile
  2.  
  3. app = FastAPI()
  4.  
  5.  
  6. @app.post("/files/")
  7. async def create_file(file: bytes = File()):
  8.     return {"file_size": len(file)}
  9.  
  10.  
  11. @app.post("/uploadfile/")
  12. async def create_upload_file(file: UploadFile):
  13.     return {"filename": file.filename}

create_file()的类型为bytes，接收到的文件内容也是bytes，数据都存在于内存中，适用于小文件。create_upload_file()的类型为UploadFile，它会在内存设置一个最大存储，超出最大存储，就会把数据转存到磁盘，适用于大文件。

UploadFile有以下属性：

  • filename，文件名，比如myimage.jpg；

  • content_type，文件类型，比如image/jpeg；

  • file，SpooledTemporaryFile实例，一个file-like对象。

UploadFile有以下方法：

  • write(data)：写数据(str或bytes)到文件；

  • read(size)：从文件读size(int)大小的bytes或character；

  • seek(offset)：定位到文件中offset(int)的位置，比如await myfile.seek(0)会定位到文件开始；

  • close()：关闭文件；

所有这些方法都是async的，需要await：

contents = await myfile.read()

不想await就使用其中的file对象：

contents = myfile.file.read()

文件可选非必传：

  1. from typing import Union
  2.  
  3. from fastapi import FastAPI, File, UploadFile
  4.  
  5. app = FastAPI()
  6.  
  7.  
  8. @app.post("/files/")
  9. async def create_file(file: Union[bytes, None] = File(default=None)):
  10.     if not file:
  11.         return {"message": "No file sent"}
  12.     else:
  13.         return {"file_size": len(file)}
  14.  
  15.  
  16. @app.post("/uploadfile/")
  17. async def create_upload_file(file: Union[UploadFile, None] = None):
  18.     if not file:
  19.         return {"message": "No upload file sent"}
  20.     else:
  21.         return {"filename": file.filename}

UploadFile的metadata：

  1. from fastapi import FastAPI, File, UploadFile
  2.  
  3. app = FastAPI()
  4.  
  5.  
  6. @app.post("/files/")
  7. async def create_file(file: bytes = File(description="A file read as bytes")):
  8.     return {"file_size": len(file)}
  9.  
  10.  
  11. @app.post("/uploadfile/")
  12. async def create_upload_file(
  13.     file: UploadFile = File(description="A file read as UploadFile"),
  14. ):
  15.     return {"filename": file.filename}

多文件上传：

  1. from typing import List
  2.  
  3. from fastapi import FastAPI, File, UploadFile
  4. from fastapi.responses import HTMLResponse
  5.  
  6. app = FastAPI()
  7.  
  8.  
  9. @app.post("/files/")
  10. async def create_files(files: List[bytes] = File()):
  11.     return {"file_sizes": [len(file) for file in files]}
  12.  
  13.  
  14. @app.post("/uploadfiles/")
  15. async def create_upload_files(files: List[UploadFile]):
  16.     return {"filenames": [file.filename for file in files]}
  17.  
  18.  
  19. @app.get("/")
  20. async def main():
  21.     content = """
  22. <body>
  23. <form action="/files/" enctype="multipart/form-data" method="post">
  24. <input name="files" type="file" multiple>
  25. <input type="submit">
  26. </form>
  27. <form action="/uploadfiles/" enctype="multipart/form-data" method="post">
  28. <input name="files" type="file" multiple>
  29. <input type="submit">
  30. </form>
  31. </body>
  32.     """
  33.     return HTMLResponse(content=content)

同时请求表单和文件

示例：

  1. from fastapi import FastAPI, File, Form, UploadFile
  2.  
  3. app = FastAPI()
  4.  
  5.  
  6. @app.post("/files/")
  7. async def create_file(
  8.     file: bytes = File(), fileb: UploadFile = File(), token: str = Form()
  9. ):
  10.     return {
  11.         "file_size": len(file),
  12.         "token": token,
  13.         "fileb_content_type": fileb.content_type,
  14.     }

错误处理

FastAPI提供了HTTPException：

  1. from fastapi import FastAPI, HTTPException
  2.  
  3. app = FastAPI()
  4.  
  5. items = {"foo": "The Foo Wrestlers"}
  6.  
  7.  
  8. @app.get("/items/{item_id}")
  9. async def read_item(item_id: str):
  10.     if item_id not in items:
  11.         raise HTTPException(status_code=404, detail="Item not found")
  12.     return {"item": items[item_id]}

HTTPException不是return而是raise的，抛出异常。

对于抛出的异常，可以使用@app.exception_handler自定义handler进行处理：

  1. from fastapi import FastAPI, Request
  2. from fastapi.responses import JSONResponse
  3.  
  4.  
  5. class UnicornException(Exception):
  6.     def __init__(self, name: str):
  7.         self.name = name
  8.  
  9.  
  10. app = FastAPI()
  11.  
  12.  
  13. @app.exception_handler(UnicornException)
  14. async def unicorn_exception_handler(request: Request, exc: UnicornException):
  15.     return JSONResponse(
  16.         status_code=418,
  17.         content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
  18.     )
  19.  
  20.  
  21. @app.get("/unicorns/{name}")
  22. async def read_unicorn(name: str):
  23.     if name == "yolo":
  24.         raise UnicornException(name=name)
  25.     return {"unicorn_name": name}

在抛出HTTPException异常时，FastAPI有很多默认的handler，比如RequestValidationError，可以使用此方法重写默认的handler：

  1. from fastapi import FastAPI, HTTPException
  2. from fastapi.exceptions import RequestValidationError
  3. from fastapi.responses import PlainTextResponse
  4. from starlette.exceptions import HTTPException as StarletteHTTPException
  5.  
  6. app = FastAPI()
  7.  
  8.  
  9. @app.exception_handler(StarletteHTTPException)
  10. async def http_exception_handler(request, exc):
  11.     return PlainTextResponse(str(exc.detail), status_code=exc.status_code)
  12.  
  13.  
  14. @app.exception_handler(RequestValidationError)
  15. async def validation_exception_handler(request, exc):
  16.     return PlainTextResponse(str(exc), status_code=400)
  17.  
  18.  
  19. @app.get("/items/{item_id}")
  20. async def read_item(item_id: int):
  21.     if item_id == 3:
  22.         raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
  23.     return {"item_id": item_id}

默认handler会返回：

  1. {
  2. "detail": [
  3.         {
  4.             "loc": [
  5.                 "path",
  6.                 "item_id"
  7.             ],
  8.             "msg": "value is not a valid integer",
  9.             "type": "type_error.integer"
  10.         }
  11.     ]
  12. }

而重写handler后会返回字符串：

  1. 1 validation error
  2. path -> item_id
  3.   value is not a valid integer (type=type_error.integer)

如果不想改动默认handler，只是补充点信息，可以导入http_exception_handler和request_validation_exception_handler：

  1. from fastapi import FastAPI, HTTPException
  2. from fastapi.exception_handlers import (
  3.     http_exception_handler,
  4.     request_validation_exception_handler,
  5. )
  6. from fastapi.exceptions import RequestValidationError
  7. from starlette.exceptions import HTTPException as StarletteHTTPException
  8.  
  9. app = FastAPI()
  10.  
  11.  
  12. @app.exception_handler(StarletteHTTPException)
  13. async def custom_http_exception_handler(request, exc):
  14.     print(f"OMG! An HTTP error!: {repr(exc)}")
  15.     return await http_exception_handler(request, exc)
  16.  
  17.  
  18. @app.exception_handler(RequestValidationError)
  19. async def validation_exception_handler(request, exc):
  20.     print(f"OMG! The client sent invalid data!: {exc}")
  21.     return await request_validation_exception_handler(request, exc)
  22.  
  23.  
  24. @app.get("/items/{item_id}")
  25. async def read_item(item_id: int):
  26.     if item_id == 3:
  27.         raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
  28.     return {"item_id": item_id}

路径操作配置

响应状态码：

  1. from typing import Set, Union
  2.  
  3. from fastapi import FastAPI, status
  4. from pydantic import BaseModel
  5.  
  6. app = FastAPI()
  7.  
  8.  
  9. class Item(BaseModel):
  10.     name: str
  11.     description: Union[str, None] = None
  12.     price: float
  13.     tax: Union[float, None] = None
  14.     tags: Set[str] = set()
  15.  
  16.  
  17. @app.post("/items/", response_model=Item, status_code=status.HTTP_201_CREATED)
  18. async def create_item(item: Item):
  19.     return item

标签：

  1. from typing import Set, Union
  2.  
  3. from fastapi import FastAPI
  4. from pydantic import BaseModel
  5.  
  6. app = FastAPI()
  7.  
  8.  
  9. class Item(BaseModel):
  10.     name: str
  11.     description: Union[str, None] = None
  12.     price: float
  13.     tax: Union[float, None] = None
  14.     tags: Set[str] = set()
  15.  
  16.  
  17. @app.post("/items/", response_model=Item, tags=["items"])
  18. async def create_item(item: Item):
  19.     return item
  20.  
  21.  
  22. @app.get("/items/", tags=["items"])
  23. async def read_items():
  24.     return [{"name": "Foo", "price": 42}]
  25.  
  26.  
  27. @app.get("/users/", tags=["users"])
  28. async def read_users():
  29.     return [{"username": "johndoe"}]

标签枚举：

  1. from enum import Enum
  2.  
  3. from fastapi import FastAPI
  4.  
  5. app = FastAPI()
  6.  
  7.  
  8. class Tags(Enum):
  9.     items = "items"
  10.     users = "users"
  11.  
  12.  
  13. @app.get("/items/", tags=[Tags.items])
  14. async def get_items():
  15.     return ["Portal gun", "Plumbus"]
  16.  
  17.  
  18. @app.get("/users/", tags=[Tags.users])
  19. async def read_users():
  20.     return ["Rick", "Morty"]

概要和描述：

  1. from typing import Set, Union
  2.  
  3. from fastapi import FastAPI
  4. from pydantic import BaseModel
  5.  
  6. app = FastAPI()
  7.  
  8.  
  9. class Item(BaseModel):
  10.     name: str
  11.     description: Union[str, None] = None
  12.     price: float
  13.     tax: Union[float, None] = None
  14.     tags: Set[str] = set()
  15.  
  16.  
  17. @app.post(
  18.     "/items/",
  19.     response_model=Item,
  20.     summary="Create an item",
  21.     description="Create an item with all the information, name, description, price, tax and a set of unique tags",
  22. )
  23. async def create_item(item: Item):
  24.     return item

文档字符串：

  1. from typing import Set, Union
  2.  
  3. from fastapi import FastAPI
  4. from pydantic import BaseModel
  5.  
  6. app = FastAPI()
  7.  
  8.  
  9. class Item(BaseModel):
  10.     name: str
  11.     description: Union[str, None] = None
  12.     price: float
  13.     tax: Union[float, None] = None
  14.     tags: Set[str] = set()
  15.  
  16.  
  17. @app.post("/items/", response_model=Item, summary="Create an item")
  18. async def create_item(item: Item):
  19.     """
  20.     Create an item with all the information:
  22.     - **name**: each item must have a name
  23.     - **description**: a long description
  24.     - **price**: required
  25.     - **tax**: if the item doesn't have tax, you can omit this
  26.     - **tags**: a set of unique tag strings for this item
  27.     """
  28.     return item

响应描述：

  1. from typing import Set, Union
  2.  
  3. from fastapi import FastAPI
  4. from pydantic import BaseModel
  5.  
  6. app = FastAPI()
  7.  
  8.  
  9. class Item(BaseModel):
  10.     name: str
  11.     description: Union[str, None] = None
  12.     price: float
  13.     tax: Union[float, None] = None
  14.     tags: Set[str] = set()
  15.  
  16.  
  17. @app.post(
  18.     "/items/",
  19.     response_model=Item,
  20.     summary="Create an item",
  21.     response_description="The created item",
  22. )
  23. async def create_item(item: Item):
  24.     """
  25.     Create an item with all the information:
  27.     - **name**: each item must have a name
  28.     - **description**: a long description
  29.     - **price**: required
  30.     - **tax**: if the item doesn't have tax, you can omit this
  31.     - **tags**: a set of unique tag strings for this item
  32.     """
  33.     return item

标记为deprecated：

  1. from fastapi import FastAPI
  2.  
  3. app = FastAPI()
  4.  
  5.  
  6. @app.get("/items/", tags=["items"])
  7. async def read_items():
  8.     return [{"name": "Foo", "price": 42}]
  9.  
  10.  
  11. @app.get("/users/", tags=["users"])
  12. async def read_users():
  13.     return [{"username": "johndoe"}]
  14.  
  15.  
  16. @app.get("/elements/", tags=["items"], deprecated=True)
  17. async def read_elements():
  18.     return [{"item_id": "Foo"}]

JSON兼容编码器

jsonable_encoder()函数的作用是把Pydantic model转换成JSON兼容的类型比如dict、list等。

  1. from datetime import datetime
  2. from typing import Union
  3.  
  4. from fastapi import FastAPI
  5. from fastapi.encoders import jsonable_encoder
  6. from pydantic import BaseModel
  7.  
  8. fake_db = {}
  9.  
  10.  
  11. class Item(BaseModel):
  12.     title: str
  13.     timestamp: datetime
  14.     description: Union[str, None] = None
  15.  
  16.  
  17. app = FastAPI()
  18.  
  19.  
  20. @app.put("/items/{id}")
  21. def update_item(id: str, item: Item):
  22.     json_compatible_item_data = jsonable_encoder(item)
  23.     fake_db[id] = json_compatible_item_data

Body-更新

使用PUT：

  1. from typing import List, Union
  2.  
  3. from fastapi import FastAPI
  4. from fastapi.encoders import jsonable_encoder
  5. from pydantic import BaseModel
  6.  
  7. app = FastAPI()
  8.  
  9.  
  10. class Item(BaseModel):
  11.     name: Union[str, None] = None
  12.     description: Union[str, None] = None
  13.     price: Union[float, None] = None
  14.     tax: float = 10.5
  15.     tags: List[str] = []
  16.  
  17.  
  18. items = {
  19.     "foo": {"name": "Foo", "price": 50.2},
  20.     "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
  21.     "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
  22. }
  23.  
  24.  
  25. @app.get("/items/{item_id}", response_model=Item)
  26. async def read_item(item_id: str):
  27.     return items[item_id]
  28.  
  29.  
  30. @app.put("/items/{item_id}", response_model=Item)
  31. async def update_item(item_id: str, item: Item):
  32.     update_item_encoded = jsonable_encoder(item)
  33.     items[item_id] = update_item_encoded
  34.     return update_item_encoded

输入数据使用了jsonable_encoder()函数转换为JSON兼容类型。

使用PATCH：

exclude_unset=True

  1. from typing import List, Union
  2.  
  3. from fastapi import FastAPI
  4. from fastapi.encoders import jsonable_encoder
  5. from pydantic import BaseModel
  6.  
  7. app = FastAPI()
  8.  
  9.  
  10. class Item(BaseModel):
  11.     name: Union[str, None] = None
  12.     description: Union[str, None] = None
  13.     price: Union[float, None] = None
  14.     tax: float = 10.5
  15.     tags: List[str] = []
  16.  
  17.  
  18. items = {
  19.     "foo": {"name": "Foo", "price": 50.2},
  20.     "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
  21.     "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
  22. }
  23.  
  24.  
  25. @app.get("/items/{item_id}", response_model=Item)
  26. async def read_item(item_id: str):
  27.     return items[item_id]
  28.  
  29.  
  30. @app.patch("/items/{item_id}", response_model=Item)
  31. async def update_item(item_id: str, item: Item):
  32.     stored_item_data = items[item_id]
  33.     stored_item_model = Item(**stored_item_data)
  34.     update_data = item.dict(exclude_unset=True)
  35.     updated_item = stored_item_model.copy(update=update_data)
  36.     items[item_id] = jsonable_encoder(updated_item)
  37.     return updated_item

.copy(update=update_data)

  1. from typing import List, Union
  2.  
  3. from fastapi import FastAPI
  4. from fastapi.encoders import jsonable_encoder
  5. from pydantic import BaseModel
  6.  
  7. app = FastAPI()
  8.  
  9.  
  10. class Item(BaseModel):
  11.     name: Union[str, None] = None
  12.     description: Union[str, None] = None
  13.     price: Union[float, None] = None
  14.     tax: float = 10.5
  15.     tags: List[str] = []
  16.  
  17.  
  18. items = {
  19.     "foo": {"name": "Foo", "price": 50.2},
  20.     "bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2},
  21.     "baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []},
  22. }
  23.  
  24.  
  25. @app.get("/items/{item_id}", response_model=Item)
  26. async def read_item(item_id: str):
  27.     return items[item_id]
  28.  
  29.  
  30. @app.patch("/items/{item_id}", response_model=Item)
  31. async def update_item(item_id: str, item: Item):
  32.     stored_item_data = items[item_id]
  33.     stored_item_model = Item(**stored_item_data)
  34.     update_data = item.dict(exclude_unset=True)
  35.     updated_item = stored_item_model.copy(update=update_data)
  36.     items[item_id] = jsonable_encoder(updated_item)
  37.     return updated_item

PUT和PATCH都可以用来部分更新，PUT用的更多。

依赖

什么是依赖注入？在FastAPI里面，你可以在路径操作函数中添加依赖的声明，然后FastAPI会自动加载这些依赖。

依赖注入的好处有：

  • 复用代码；
  • 复用数据库连接；
  • 增强安全、认证、角色；
  • 等等等；

依赖注入示例：

  1. from typing import Union
  2.  
  3. from fastapi import Depends, FastAPI
  4.  
  5. app = FastAPI()
  6.  
  7.  
  8. async def common_parameters(
  9.     q: Union[str, None] = None, skip: int = 0, limit: int = 100
  10. ):
  11.     return {"q": q, "skip": skip, "limit": limit}
  12.  
  13.  
  14. @app.get("/items/")
  15. async def read_items(commons: dict = Depends(common_parameters)):
  16.     return commons
  17.  
  18.  
  19. @app.get("/users/")
  20. async def read_users(commons: dict = Depends(common_parameters)):
  21.     return commons
  • common_parameters()函数是个简单的依赖；
  • Depends引入依赖；

FastAPI就会自动调用common_parameters()函数并把结果返回给commons，而无需任何其他代码。

依赖也可以使用class，把common_parameters()函数改为CommonQueryParams类：

  1. from typing import Union
  2.  
  3. from fastapi import Depends, FastAPI
  4.  
  5. app = FastAPI()
  6.  
  7.  
  8. fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
  9.  
  10.  
  11. class CommonQueryParams:
  12.     def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100):
  13.         self.q = q
  14.         self.skip = skip
  15.         self.limit = limit
  16.  
  17.  
  18. @app.get("/items/")
  19. async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
  20.     response = {}
  21.     if commons.q:
  22.         response.update({"q": commons.q})
  23.     items = fake_items_db[commons.skip : commons.skip + commons.limit]
  24.     response.update({"items": items})
  25.     return response

Depends会创建一个CommonQueryParams的实例然后赋值给commons。

更一步简化，只写1次CommonQueryParams：

commons: CommonQueryParams = Depends()

Depends()里面的CommonQueryParams可以省略掉。

FastAPI支持子依赖，也就是Depends嵌套：

  1. from typing import Union
  2.  
  3. from fastapi import Cookie, Depends, FastAPI
  4.  
  5. app = FastAPI()
  6.  
  7.  
  8. def query_extractor(q: Union[str, None] = None):
  9.     return q
  10.  
  11.  
  12. def query_or_cookie_extractor(
  13.     q: str = Depends(query_extractor),
  14.     last_query: Union[str, None] = Cookie(default=None),
  15. ):
  16.     if not q:
  17.         return last_query
  18.     return q
  19.  
  20.  
  21. @app.get("/items/")
  22. async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):
  23.     return {"q_or_cookie": query_or_default}

如果使用同一个依赖多次，FastAPI默认会只注入一次。可以按以下设置让FastAPI注入多次：

  1. async def needy_dependency(fresh_value: str = Depends(get_value, use_cache=False)):
  2.     return {"fresh_value": fresh_value}

多个依赖可以用dependencies的list：

  1. from fastapi import Depends, FastAPI, Header, HTTPException
  2.  
  3. app = FastAPI()
  4.  
  5.  
  6. async def verify_token(x_token: str = Header()):
  7.     if x_token != "fake-super-secret-token":
  8.         raise HTTPException(status_code=400, detail="X-Token header invalid")
  9.  
  10.  
  11. async def verify_key(x_key: str = Header()):
  12.     if x_key != "fake-super-secret-key":
  13.         raise HTTPException(status_code=400, detail="X-Key header invalid")
  14.     return x_key
  15.  
  16.  
  17. @app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
  18. async def read_items():
  19.     return [{"item": "Foo"}, {"item": "Bar"}]

如果给FastAPI的构造函数传入dependencies，那么就是全局依赖：

  1. from fastapi import Depends, FastAPI, Header, HTTPException
  2.  
  3.  
  4. async def verify_token(x_token: str = Header()):
  5.     if x_token != "fake-super-secret-token":
  6.         raise HTTPException(status_code=400, detail="X-Token header invalid")
  7.  
  8.  
  9. async def verify_key(x_key: str = Header()):
  10.     if x_key != "fake-super-secret-key":
  11.         raise HTTPException(status_code=400, detail="X-Key header invalid")
  12.     return x_key
  13.  
  14.  
  15. app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])
  16.  
  17.  
  18. @app.get("/items/")
  19. async def read_items():
  20.     return [{"item": "Portal Gun"}, {"item": "Plumbus"}]
  21.  
  22.  
  23. @app.get("/users/")
  24. async def read_users():
  25.     return [{"username": "Rick"}, {"username": "Morty"}]

如果在依赖函数中使用yield，它后面的代码就相当于teardown，这点用法跟pytest的fixture类似：

  1. async def get_db():
  2.     db = DBSession()
  3.     try:
  4.         yield db
  5.     finally:
  6.         db.close()

另外，借助yield和with可以创建一个上下文管理器(实现__enter__和__exit__)：

  1. class MySuperContextManager:
  2.     def __init__(self):
  3.         self.db = DBSession()
  4.  
  5.     def __enter__(self):
  6.         return self.db
  7.  
  8.     def __exit__(self, exc_type, exc_value, traceback):
  9.         self.db.close()
  10.  
  11.  
  12. async def get_db():
  13.     with MySuperContextManager() as db:
  14.         yield db

安全

FastAPI支持OAuth2协议：

  1. from fastapi import Depends, FastAPI
  2. from fastapi.security import OAuth2PasswordBearer
  3.  
  4. app = FastAPI()
  5.  
  6. oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
  7.  
  8.  
  9. @app.get("/items/")
  10. async def read_items(token: str = Depends(oauth2_scheme)):
  11.     return {"token": token}

备注：需要提前安装 pip install python-multipart，因为OAuth2使用表单来发送username和password。虽然这个接口已经加上鉴权了。但这些入参都没有生效，因为我们还没有添加相应的处理代码。为了让鉴权实际生效，我们继续添加代码：

  1. from typing import Union
  2.  
  3. from fastapi import Depends, FastAPI
  4. from fastapi.security import OAuth2PasswordBearer
  5. from pydantic import BaseModel
  6.  
  7. app = FastAPI()
  8.  
  9. oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
  10.  
  11.  
  12. class User(BaseModel):
  13.     username: str
  14.     email: Union[str, None] = None
  15.     full_name: Union[str, None] = None
  16.     disabled: Union[bool, None] = None
  17.  
  18.  
  19. def fake_decode_token(token):
  20.     return User(
  21.         username=token + "fakedecoded", email="john@example.com", full_name="John Doe"
  22.     )
  23.  
  24.  
  25. async def get_current_user(token: str = Depends(oauth2_scheme)):
  26.     user = fake_decode_token(token)
  27.     return user
  28.  
  29.  
  30. @app.get("/users/me")
  31. async def read_users_me(current_user: User = Depends(get_current_user)):
  32.     return current_user
  1. 定义模型User；
  2. 创建依赖get_current_user；
  3. fake_decode_token接收token，返回模拟的假用户；
  4. read_users_me注入依赖；

然后实现username和password：

  1. from typing import Union
  2.  
  3. from fastapi import Depends, FastAPI, HTTPException, status
  4. from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
  5. from pydantic import BaseModel
  6.  
  7. fake_users_db = {
  8.     "johndoe": {
  9.         "username": "johndoe",
  10.         "full_name": "John Doe",
  11.         "email": "johndoe@example.com",
  12.         "hashed_password": "fakehashedsecret",
  13.         "disabled": False,
  14.     },
  15.     "alice": {
  16.         "username": "alice",
  17.         "full_name": "Alice Wonderson",
  18.         "email": "alice@example.com",
  19.         "hashed_password": "fakehashedsecret2",
  20.         "disabled": True,
  21.     },
  22. }
  23.  
  24. app = FastAPI()
  25.  
  26.  
  27. def fake_hash_password(password: str):
  28.     return "fakehashed" + password
  29.  
  30.  
  31. oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
  32.  
  33.  
  34. class User(BaseModel):
  35.     username: str
  36.     email: Union[str, None] = None
  37.     full_name: Union[str, None] = None
  38.     disabled: Union[bool, None] = None
  39.  
  40.  
  41. class UserInDB(User):
  42.     hashed_password: str
  43.  
  44.  
  45. def get_user(db, username: str):
  46.     if username in db:
  47.         user_dict = db[username]
  48.         return UserInDB(**user_dict)
  49.  
  50.  
  51. def fake_decode_token(token):
  52.     # This doesn't provide any security at all
  53.     # Check the next version
  54.     user = get_user(fake_users_db, token)
  55.     return user
  56.  
  57.  
  58. async def get_current_user(token: str = Depends(oauth2_scheme)):
  59.     user = fake_decode_token(token)
  60.     if not user:
  61.         raise HTTPException(
  62.             status_code=status.HTTP_401_UNAUTHORIZED,
  63.             detail="Invalid authentication credentials",
  64.             headers={"WWW-Authenticate": "Bearer"},
  65.         )
  66.     return user
  67.  
  68.  
  69. async def get_current_active_user(current_user: User = Depends(get_current_user)):
  70.     if current_user.disabled:
  71.         raise HTTPException(status_code=400, detail="Inactive user")
  72.     return current_user
  73.  
  74.  
  75. @app.post("/token")
  76. async def login(form_data: OAuth2PasswordRequestForm = Depends()):
  77.     user_dict = fake_users_db.get(form_data.username)
  78.     if not user_dict:
  79.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  80.     user = UserInDB(**user_dict)
  81.     hashed_password = fake_hash_password(form_data.password)
  82.     if not hashed_password == user.hashed_password:
  83.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  84.  
  85.     return {"access_token": user.username, "token_type": "bearer"}
  86.  
  87.  
  88. @app.get("/users/me")
  89. async def read_users_me(current_user: User = Depends(get_current_active_user)):
  90.     return current_user

现在就可以测试一下了http://127.0.0.1:8000/docs：

img

授权以后：

img

访问/users/me会返回：

  1. {
  2.   "username": "johndoe",
  3.   "email": "johndoe@example.com",
  4.   "full_name": "John Doe",
  5.   "disabled": false,
  6.   "hashed_password": "fakehashedsecret"
  7. }

如果logout再访问会出现：

  1. {
  2.   "detail": "Not authenticated"
  3. }

输入错误的用户会出现：

  1. {
  2.   "detail": "Inactive user"
  3. }

如果想使用JWT，那么先安装python-jose。为了给密码加密，需要安装passlib。

示例代码：

  1. from datetime import datetime, timedelta
  2. from typing import Union
  3.  
  4. from fastapi import Depends, FastAPI, HTTPException, status
  5. from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
  6. from jose import JWTError, jwt
  7. from passlib.context import CryptContext
  8. from pydantic import BaseModel
  9.  
  10. # to get a string like this run:
  11. # openssl rand -hex 32
  12. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  13. ALGORITHM = "HS256"
  14. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  15.  
  16.  
  17. fake_users_db = {
  18.     "johndoe": {
  19.         "username": "johndoe",
  20.         "full_name": "John Doe",
  21.         "email": "johndoe@example.com",
  22.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  23.         "disabled": False,
  24.     }
  25. }
  26.  
  27.  
  28. class Token(BaseModel):
  29.     access_token: str
  30.     token_type: str
  31.  
  32.  
  33. class TokenData(BaseModel):
  34.     username: Union[str, None] = None
  35.  
  36.  
  37. class User(BaseModel):
  38.     username: str
  39.     email: Union[str, None] = None
  40.     full_name: Union[str, None] = None
  41.     disabled: Union[bool, None] = None
  42.  
  43.  
  44. class UserInDB(User):
  45.     hashed_password: str
  46.  
  47.  
  48. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  49.  
  50. oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
  51.  
  52. app = FastAPI()
  53.  
  54.  
  55. def verify_password(plain_password, hashed_password):
  56.     return pwd_context.verify(plain_password, hashed_password)
  57.  
  58.  
  59. def get_password_hash(password):
  60.     return pwd_context.hash(password)
  61.  
  62.  
  63. def get_user(db, username: str):
  64.     if username in db:
  65.         user_dict = db[username]
  66.         return UserInDB(**user_dict)
  67.  
  68.  
  69. def authenticate_user(fake_db, username: str, password: str):
  70.     user = get_user(fake_db, username)
  71.     if not user:
  72.         return False
  73.     if not verify_password(password, user.hashed_password):
  74.         return False
  75.     return user
  76.  
  77.  
  78. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  79.     to_encode = data.copy()
  80.     if expires_delta:
  81.         expire = datetime.utcnow() + expires_delta
  82.     else:
  83.         expire = datetime.utcnow() + timedelta(minutes=15)
  84.     to_encode.update({"exp": expire})
  85.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  86.     return encoded_jwt
  87.  
  88.  
  89. async def get_current_user(token: str = Depends(oauth2_scheme)):
  90.     credentials_exception = HTTPException(
  91.         status_code=status.HTTP_401_UNAUTHORIZED,
  92.         detail="Could not validate credentials",
  93.         headers={"WWW-Authenticate": "Bearer"},
  94.     )
  95.     try:
  96.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  97.         username: str = payload.get("sub")
  98.         if username is None:
  99.             raise credentials_exception
  100.         token_data = TokenData(username=username)
  101.     except JWTError:
  102.         raise credentials_exception
  103.     user = get_user(fake_users_db, username=token_data.username)
  104.     if user is None:
  105.         raise credentials_exception
  106.     return user
  107.  
  108.  
  109. async def get_current_active_user(current_user: User = Depends(get_current_user)):
  110.     if current_user.disabled:
  111.         raise HTTPException(status_code=400, detail="Inactive user")
  112.     return current_user
  113.  
  114.  
  115. @app.post("/token", response_model=Token)
  116. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
  117.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  118.     if not user:
  119.         raise HTTPException(
  120.             status_code=status.HTTP_401_UNAUTHORIZED,
  121.             detail="Incorrect username or password",
  122.             headers={"WWW-Authenticate": "Bearer"},
  123.         )
  124.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  125.     access_token = create_access_token(
  126.         data={"sub": user.username}, expires_delta=access_token_expires
  127.     )
  128.     return {"access_token": access_token, "token_type": "bearer"}
  129.  
  130.  
  131. @app.get("/users/me/", response_model=User)
  132. async def read_users_me(current_user: User = Depends(get_current_active_user)):
  133.     return current_user
  134.  
  135.  
  136. @app.get("/users/me/items/")
  137. async def read_own_items(current_user: User = Depends(get_current_active_user)):
  138.     return [{"item_id": "Foo", "owner": current_user.username}]

其中的SECRET_KEY通过openssl生成：

openssl rand -hex 32

中间件

FastAPI这里的中间件，指的是一个函数，它在请求处理前被调用，在响应返回前调用。有点类似于Spring的过滤器filter。

创建中间件：

  1. import time
  2.  
  3. from fastapi import FastAPI, Request
  4.  
  5. app = FastAPI()
  6.  
  7.  
  8. @app.middleware("http")
  9. async def add_process_time_header(request: Request, call_next):
  10.     start_time = time.time()
  11.     response = await call_next(request)
  12.     process_time = time.time() - start_time
  13.     response.headers["X-Process-Time"] = str(process_time)
  14.     return response

CORS

Cross-Origin Resource Sharing，跨域访问。

同域包括协议、域名、端口，以下均是不同域：

  • http://localhost
  • https://localhost
  • http://localhost:8080

使用CORSMiddleware可以实现跨域访问：

  1. from fastapi import FastAPI
  2. from fastapi.middleware.cors import CORSMiddleware
  3.  
  4. app = FastAPI()
  5.  
  6. origins = [
  7.     "http://localhost.tiangolo.com",
  8.     "https://localhost.tiangolo.com",
  9.     "http://localhost",
  10.     "http://localhost:8080",
  11. ]
  12.  
  13. app.add_middleware(
  14.     CORSMiddleware,
  15.     allow_origins=origins,
  16.     allow_credentials=True,
  17.     allow_methods=["*"],
  18.     allow_headers=["*"],
  19. )
  20.  
  21.  
  22. @app.get("/")
  23. async def main():
  24.     return {"message": "Hello World"}
  • allow_origins，允许域名，[*]代表所有；
  • allow_origin_regex，允许域名的正则匹配；
  • allow_methods，允许请求方法，[*]代表所有；
  • allow_headers，允许请求头，[*]代表所有；
  • allow_credentials，跨域访问时是否需要cookie，默认False，设置为True时allow_origins不能设置为[*]；
  • expose_headers，暴露给浏览器的响应头，默认[]；
  • max_age，浏览器最大缓存CORS 响应的时间，默认60s；

SQL关系型数据库

官方教程使用的是SQLAlchemy。

示例：

  1. .
  2. └── sql_app
  3.     ├── __init__.py
  4.     ├── crud.py
  5.     ├── database.py
  6.     ├── main.py
  7.     ├── models.py
  8.     └── schemas.py

安装：

pip install sqlalchemy

创建数据库：

  1. from sqlalchemy import create_engine
  2. from sqlalchemy.ext.declarative import declarative_base
  3. from sqlalchemy.orm import sessionmaker
  4.  
  5. SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
  6. # SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"
  7.  
  8. engine = create_engine(
  9.     SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
  10. )
  11. SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
  12.  
  13. Base = declarative_base()

创建数据库模型：

  1. from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
  2. from sqlalchemy.orm import relationship
  3.  
  4. from .database import Base
  5.  
  6.  
  7. class User(Base):
  8.     __tablename__ = "users"
  9.  
  10.     id = Column(Integer, primary_key=True, index=True)
  11.     email = Column(String, unique=True, index=True)
  12.     hashed_password = Column(String)
  13.     is_active = Column(Boolean, default=True)
  14.  
  15.     items = relationship("Item", back_populates="owner")
  16.  
  17.  
  18. class Item(Base):
  19.     __tablename__ = "items"
  20.  
  21.     id = Column(Integer, primary_key=True, index=True)
  22.     title = Column(String, index=True)
  23.     description = Column(String, index=True)
  24.     owner_id = Column(Integer, ForeignKey("users.id"))
  25.  
  26.     owner = relationship("User", back_populates="items")

创建Pydantic模型：

  1. from typing import List, Union
  2.  
  3. from pydantic import BaseModel
  4.  
  5.  
  6. class ItemBase(BaseModel):
  7.     title: str
  8.     description: Union[str, None] = None
  9.  
  10.  
  11. class ItemCreate(ItemBase):
  12.     pass
  13.  
  14.  
  15. class Item(ItemBase):
  16.     id: int
  17.     owner_id: int
  18.  
  19.     class Config:
  20.         orm_mode = True
  21.  
  22.  
  23. class UserBase(BaseModel):
  24.     email: str
  25.  
  26.  
  27. class UserCreate(UserBase):
  28.     password: str
  29.  
  30.  
  31. class User(UserBase):
  32.     id: int
  33.     is_active: bool
  34.     items: List[Item] = []
  35.  
  36.     class Config:
  37.         orm_mode = True

注意，SQLAlchemy使用=赋值，Pydantic使用:赋值。

增删改查：

  1. from sqlalchemy.orm import Session
  2.  
  3. from . import models, schemas
  4.  
  5.  
  6. def get_user(db: Session, user_id: int):
  7.     return db.query(models.User).filter(models.User.id == user_id).first()
  8.  
  9.  
  10. def get_user_by_email(db: Session, email: str):
  11.     return db.query(models.User).filter(models.User.email == email).first()
  12.  
  13.  
  14. def get_users(db: Session, skip: int = 0, limit: int = 100):
  15.     return db.query(models.User).offset(skip).limit(limit).all()
  16.  
  17.  
  18. def create_user(db: Session, user: schemas.UserCreate):
  19.     fake_hashed_password = user.password + "notreallyhashed"
  20.     db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
  21.     db.add(db_user)
  22.     db.commit()
  23.     db.refresh(db_user)
  24.     return db_user
  25.  
  26.  
  27. def get_items(db: Session, skip: int = 0, limit: int = 100):
  28.     return db.query(models.Item).offset(skip).limit(limit).all()
  29.  
  30.  
  31. def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
  32.     db_item = models.Item(**item.dict(), owner_id=user_id)
  33.     db.add(db_item)
  34.     db.commit()
  35.     db.refresh(db_item)
  36.     return db_item

主程序：

  1. from typing import List
  2.  
  3. from fastapi import Depends, FastAPI, HTTPException
  4. from sqlalchemy.orm import Session
  5.  
  6. from . import crud, models, schemas
  7. from .database import SessionLocal, engine
  8.  
  9. models.Base.metadata.create_all(bind=engine)
  10.  
  11. app = FastAPI()
  12.  
  13.  
  14. # Dependency
  15. def get_db():
  16.     db = SessionLocal()
  17.     try:
  18.         yield db
  19.     finally:
  20.         db.close()
  21.  
  22.  
  23. @app.post("/users/", response_model=schemas.User)
  24. def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
  25.     db_user = crud.get_user_by_email(db, email=user.email)
  26.     if db_user:
  27.         raise HTTPException(status_code=400, detail="Email already registered")
  28.     return crud.create_user(db=db, user=user)
  29.  
  30.  
  31. @app.get("/users/", response_model=List[schemas.User])
  32. def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
  33.     users = crud.get_users(db, skip=skip, limit=limit)
  34.     return users
  35.  
  36.  
  37. @app.get("/users/{user_id}", response_model=schemas.User)
  38. def read_user(user_id: int, db: Session = Depends(get_db)):
  39.     db_user = crud.get_user(db, user_id=user_id)
  40.     if db_user is None:
  41.         raise HTTPException(status_code=404, detail="User not found")
  42.     return db_user
  43.  
  44.  
  45. @app.post("/users/{user_id}/items/", response_model=schemas.Item)
  46. def create_item_for_user(
  47.     user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
  48. ):
  49.     return crud.create_user_item(db=db, item=item, user_id=user_id)
  50.  
  51.  
  52. @app.get("/items/", response_model=List[schemas.Item])
  53. def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
  54.     items = crud.get_items(db, skip=skip, limit=limit)
  55.     return items

大应用-多文件

示例目录结构：

  1. .
  2. ├── app                  # "app" is a Python package
  3. │   ├── __init__.py      # this file makes "app" a "Python package"
  4. │   ├── main.py          # "main" module, e.g. import app.main
  5. │   ├── dependencies.py  # "dependencies" module, e.g. import app.dependencies
  6. │   └── routers          # "routers" is a "Python subpackage"
  7. │   │   ├── __init__.py  # makes "routers" a "Python subpackage"
  8. │   │   ├── items.py     # "items" submodule, e.g. import app.routers.items
  9. │   │   └── users.py     # "users" submodule, e.g. import app.routers.users
  10. │   └── internal         # "internal" is a "Python subpackage"
  11. │       ├── __init__.py  # makes "internal" a "Python subpackage"
  12. │       └── admin.py     # "admin" submodule, e.g. import app.internal.admin

APIRouter用于定义子模块的路由：

  1. from fastapi import APIRouter
  2.  
  3. router = APIRouter()
  4.  
  5.  
  6. @router.get("/users/", tags=["users"])
  7. async def read_users():
  8.     return [{"username": "Rick"}, {"username": "Morty"}]
  9.  
  10.  
  11. @router.get("/users/me", tags=["users"])
  12. async def read_user_me():
  13.     return {"username": "fakecurrentuser"}
  14.  
  15.  
  16. @router.get("/users/{username}", tags=["users"])
  17. async def read_user(username: str):
  18.     return {"username": username}
  1. from fastapi import APIRouter, Depends, HTTPException
  2.  
  3. from ..dependencies import get_token_header
  4.  
  5. router = APIRouter(
  6.     prefix="/items",
  7.     tags=["items"],
  8.     dependencies=[Depends(get_token_header)],
  9.     responses={404: {"description": "Not found"}},
  10. )
  11.  
  12.  
  13. fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}}
  14.  
  15.  
  16. @router.get("/")
  17. async def read_items():
  18.     return fake_items_db
  19.  
  20.  
  21. @router.get("/{item_id}")
  22. async def read_item(item_id: str):
  23.     if item_id not in fake_items_db:
  24.         raise HTTPException(status_code=404, detail="Item not found")
  25.     return {"name": fake_items_db[item_id]["name"], "item_id": item_id}
  26.  
  27.  
  28. @router.put(
  29.     "/{item_id}",
  30.     tags=["custom"],
  31.     responses={403: {"description": "Operation forbidden"}},
  32. )
  33. async def update_item(item_id: str):
  34.     if item_id != "plumbus":
  35.         raise HTTPException(
  36.             status_code=403, detail="You can only update the item: plumbus"
  37.         )
  38.     return {"item_id": item_id, "name": "The great Plumbus"}

在主程序中引入子模块路由：

  1. from fastapi import Depends, FastAPI
  2.  
  3. from .dependencies import get_query_token, get_token_header
  4. from .internal import admin
  5. from .routers import items, users
  6.  
  7. app = FastAPI(dependencies=[Depends(get_query_token)])
  8.  
  9.  
  10. app.include_router(users.router)
  11. app.include_router(items.router)
  12. app.include_router(
  13.     admin.router,
  14.     prefix="/admin",
  15.     tags=["admin"],
  16.     dependencies=[Depends(get_token_header)],
  17.     responses={418: {"description": "I'm a teapot"}},
  18. )
  19.  
  20.  
  21. @app.get("/")
  22. async def root():
  23.     return {"message": "Hello Bigger Applications!"}

后台任务

使用BackgroundTasks定义后台任务：

  1. from fastapi import BackgroundTasks, FastAPI
  2.  
  3. app = FastAPI()
  4.  
  5.  
  6. def write_notification(email: str, message=""):
  7.     with open("log.txt", mode="w") as email_file:
  8.         content = f"notification for {email}: {message}"
  9.         email_file.write(content)
  10.  
  11.  
  12. @app.post("/send-notification/{email}")
  13. async def send_notification(email: str, background_tasks: BackgroundTasks):
  14.     background_tasks.add_task(write_notification, email, message="some notification")
  15.     return {"message": "Notification sent in the background"}

BackgroundTasks也能支持依赖注入：

  1. from typing import Union
  2.  
  3. from fastapi import BackgroundTasks, Depends, FastAPI
  4.  
  5. app = FastAPI()
  6.  
  7.  
  8. def write_log(message: str):
  9.     with open("log.txt", mode="a") as log:
  10.         log.write(message)
  11.  
  12.  
  13. def get_query(background_tasks: BackgroundTasks, q: Union[str, None] = None):
  14.     if q:
  15.         message = f"found query: {q}\n"
  16.         background_tasks.add_task(write_log, message)
  17.     return q
  18.  
  19.  
  20. @app.post("/send-notification/{email}")
  21. async def send_notification(
  22.     email: str, background_tasks: BackgroundTasks, q: str = Depends(get_query)
  23. ):
  24.     message = f"message to {email}\n"
  25.     background_tasks.add_task(write_log, message)
  26.     return {"message": "Message sent"}

元数据和文档URL

设置应用元数据：

  1. from fastapi import FastAPI
  2.  
  3. description = """
  4. ChimichangApp API helps you do awesome stuff. 
  6. ## Items
  8. You can **read items**.
  10. ## Users
  12. You will be able to:
  14. * **Create users** (_not implemented_).
  15. * **Read users** (_not implemented_).
  16. """
  17.  
  18. app = FastAPI(
  19.     title="ChimichangApp",
  20.     description=description,
  21.     version="0.0.1",
  22.     terms_of_service="http://example.com/terms/",
  23.     contact={
  24.         "name": "Deadpoolio the Amazing",
  25.         "url": "http://x-force.example.com/contact/",
  26.         "email": "dp@x-force.example.com",
  27.     },
  28.     license_info={
  29.         "name": "Apache 2.0",
  30.         "url": "https://www.apache.org/licenses/LICENSE-2.0.html",
  31.     },
  32. )
  33.  
  34.  
  35. @app.get("/items/")
  36. async def read_items():
  37.     return [{"name": "Katana"}]

效果：

img

设置tag元数据：

  1. from fastapi import FastAPI
  2.  
  3. tags_metadata = [
  4.     {
  5.         "name": "users",
  6.         "description": "Operations with users. The **login** logic is also here.",
  7.     },
  8.     {
  9.         "name": "items",
  10.         "description": "Manage items. So _fancy_ they have their own docs.",
  11.         "externalDocs": {
  12.             "description": "Items external docs",
  13.             "url": "https://fastapi.tiangolo.com/",
  14.         },
  15.     },
  16. ]
  17.  
  18. app = FastAPI(openapi_tags=tags_metadata)
  19.  
  20.  
  21. @app.get("/users/", tags=["users"])
  22. async def get_users():
  23.     return [{"name": "Harry"}, {"name": "Ron"}]
  24.  
  25.  
  26. @app.get("/items/", tags=["items"])
  27. async def get_items():
  28.     return [{"name": "wand"}, {"name": "flying broom"}]

添加tag：

  1. from fastapi import FastAPI
  2.  
  3. tags_metadata = [
  4.     {
  5.         "name": "users",
  6.         "description": "Operations with users. The **login** logic is also here.",
  7.     },
  8.     {
  9.         "name": "items",
  10.         "description": "Manage items. So _fancy_ they have their own docs.",
  11.         "externalDocs": {
  12.             "description": "Items external docs",
  13.             "url": "https://fastapi.tiangolo.com/",
  14.         },
  15.     },
  16. ]
  17.  
  18. app = FastAPI(openapi_tags=tags_metadata)
  19.  
  20.  
  21. @app.get("/users/", tags=["users"])
  22. async def get_users():
  23.     return [{"name": "Harry"}, {"name": "Ron"}]
  24.  
  25.  
  26. @app.get("/items/", tags=["items"])
  27. async def get_items():
  28.     return [{"name": "wand"}, {"name": "flying broom"}]

效果：

img

OpenAPI的URL默认是/openapi.json，设置/api/v1/openapi.json：

  1. from fastapi import FastAPI
  2.  
  3. app = FastAPI(openapi_url="/api/v1/openapi.json")
  4.  
  5.  
  6. @app.get("/items/")
  7. async def read_items():
  8.     return [{"name": "Foo"}]

文档的URL默认是/docs，设置为/documentation：

  1. from fastapi import FastAPI
  2.  
  3. app = FastAPI(docs_url="/documentation", redoc_url=None)
  4.  
  5.  
  6. @app.get("/items/")
  7. async def read_items():
  8.     return [{"name": "Foo"}]

静态文件

使用StaticFiles：

  1. from fastapi import FastAPI
  2. from fastapi.staticfiles import StaticFiles
  3.  
  4. app = FastAPI()
  5.  
  6. app.mount("/static", StaticFiles(directory="static"), name="static")

FastAPI会自动挂载静态文件。

单元测试

使用pytest和TestClient：

  1. from fastapi import FastAPI
  2. from fastapi.testclient import TestClient
  3.  
  4. app = FastAPI()
  5.  
  6.  
  7. @app.get("/")
  8. async def read_main():
  9.     return {"msg": "Hello World"}
  10.  
  11.  
  12. client = TestClient(app)
  13.  
  14.  
  15. def test_read_main():
  16.     response = client.get("/")
  17.     assert response.status_code == 200
  18.     assert response.json() == {"msg": "Hello World"}

单元测试文件拆出来：

  1. .
  2. ├── app
  3. │   ├── __init__.py
  4. │   ├── main.py
  5. │   └── test_main.py
  1. from fastapi import FastAPI
  2.  
  3. app = FastAPI()
  4.  
  5.  
  6. @app.get("/")
  7. async def read_main():
  8.     return {"msg": "Hello World"}
  1. from fastapi.testclient import TestClient
  2.  
  3. from .main import app
  4.  
  5. client = TestClient(app)
  6.  
  7.  
  8. def test_read_main():
  9.     response = client.get("/")
  10.     assert response.status_code == 200
  11.     assert response.json() == {"msg": "Hello World"}

调试

通常借助PyCharm打断点调试：

img

参考资料：

官方教程 https://fastapi.tiangolo.com/tutorial/

示例项目 https://fastapi.tiangolo.com/project-generation/

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/菜鸟追梦旅行/article/detail/416495?site
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号