天天看点

FastAPI 快速上手,FastAPI程序模版FastAPI快速上手

FastAPI快速上手

本文主要是分享如何快速上手fastAPI,如果读者已经有flask sanic django等框架的使用经验,那么本文应该没有太大的阅读难度。官方文档,本文会尽可能避免将文章写成官方文档。

有关fastAPI 的安装直接看官方文档

阶段1

项目结构

└── fastAPI_app
    └── run.py
           

code

# fastAPI_app/run.py
from fastapi import FastAPI

fastApp = FastAPI()

@fastApp.get('/test')
async def get():
    return {"msg": "hello"}

# uvicorn app:app --reload
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(fastApp, host='127.0.0.1', port=9998)
           

与flask很像,特别像,fastAPI还提供了一个交互式的API文档,执行

python run.py

,然后在浏览器输入http://127.0.0.1:9998/docs就可以看到

FastAPI 快速上手,FastAPI程序模版FastAPI快速上手

点击 这个接口进入

FastAPI 快速上手,FastAPI程序模版FastAPI快速上手

再点击try it out

FastAPI 快速上手,FastAPI程序模版FastAPI快速上手

现在可以点解 excute,就可以测试该接口了,有用过Swagger的人应该都知道,没用过的可以自己去探索一下。

请求体

有关请求参数、查询参数等,这里不浪费口舌了,这里就说明一下 post方法传入json格式的参数。官方文档

#fastAPI_app/run.py
from fastapi import FastAPI
from pydantic import BaseModel

from typing import Optional

fastApp = FastAPI()

@fastApp.get('/test')
async def get():
    return {"msg": "hello"}

# declare a request body
class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

# 如果仅仅是用来python类型声明,fastAPI 会将请求体作为JSON读取
@fastApp.post("/test/item_query/")
async def create_item(item: Item):
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

# uvicorn app:app --reload
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(fastApp, host='127.0.0.1', port=9998)
           

执行文件,接着在Swagger中测试,使用pydantic可以直接约束请求体,fastAPI会把你定义的model解析成JSON并在Swagger中直接体现出来。当然了我们也可以使用Body,有关查询参数、路径参数、header参数、cookie参数等,看文档。

阶段2

项目结构

└── fastAPI_app
    ├── app
    │   ├── __init__.py
    │   └── routers
    │       ├── __init__.py
    │       └── test.py
    └── run.py
           

code

# fastAPI_app/run.py
from app import create_app

fastApp = create_app()

# uvicorn app:app --reload
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(fastApp, host='127.0.0.1', port=9998)
           
#fastAPI_app/app/__init__.py
from fastapi import FastAPI

from app.routers import testRouter

def add_router(app: FastAPI):
    app.include_router(testRouter)

def create_app():
    app = FastAPI()
    add_router(app)

    return app
           
#fastAPI_app/app/routers/test.py
from fastapi import APIRouter
from pydantic import BaseModel

from typing import Optional


testRouter = APIRouter()

@testRouter.get('/test')
async def get():
    return {"msg": "hello"}

# declare a request body
class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None


@testRouter.post("/test/item_query/")
async def create_item(item: Item):
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict
           
#fastAPI_app/app/routers/__init__.py
from .test import testRouter
           

相比于阶段1,功能没有任何变化,只是改变了项目结构,同时引入了新的FastAPI路由使用方法,如果你使用过flask的blueprint,这个基本上也就没多大难度了,文档链接,我在routers目录下新建了test.py文件,并在这里创建了一个路由实例,最后将这个路由实例挂载在app实例上。这样做的好处就是,代码干净。

视图模型

之前在routers中的test.py中创建了model,可以把这个model迁移出来

└── fastAPI_app
    ├── app
    │   ├── __init__.py
    │   ├── routers
    │   │   ├── __init__.py
    │   │   └── test.py
    │   └── view_models
    │       ├── __init__.py
    │       └── test.py
    └── run.py
           
#fastAPI_app/app/routers/test.py
from fastapi import APIRouter
from app.view_models.test import Item

testRouter = APIRouter()

@testRouter.get('/test')
async def get():
    return {"msg": "hello"}

@testRouter.post("/test/item_query/")
async def create_item(item: Item):
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict
           
#fastAPI_app/app/view_models/test.py
from pydantic import BaseModel
from typing import Optional

# declare a request body
class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
           

现在,这个项目已经有点眉目了,使用视图模型和路由共同解析用户传来的参数,并在路由中处理参数并返回结果,接着就是要操作数据库了,我们肯定不希望在路由中操作数据库。

阶段3

├── FastAPI_base
│   ├── app
│   │   ├── __init__.py
│   │   ├── controller
│   │   │   ├── __init__.py
│   │   │   └── test.py
│   │   ├── data
│   │   │   └── test.db
│   │   ├── models
│   │   │   ├── __init__.py
│   │   │   └── test.py
│   │   ├── routers
│   │   │   ├── __init__.py
│   │   │   └── test.py
│   │   └── view_models
│   │       ├── __init__.py
│   │       └── test.py
│   ├── requirements.txt
│   └── run.py
           

此处新添加了controller和models,models用来定义ORM模型,controller中定义增删改查,当用户的请求需要操作数据库时,会在路由中调用controller,简单的MVC架构。SQL FastAPI

,在官方文档中使用的是 SQLAlchemy,所以,如果你有一套属于自己风格的db操作方法,可以跳过了。具体的代码可以在笔者的项目中查看FastAPI_base,笔者的目标是根据官方文档完成一个FastAPI程序模版,类似于flask_base那样,后续会尝试添加 async_sql, GraphQL, WebSocket等,也希望有人可以一起参与其中,共同学习。