美文网首页
FastAPI for Fun

FastAPI for Fun

作者: 黑夜的眸 | 来源:发表于2020-03-15 16:27 被阅读0次

    本文主要针对使用FastAPI构建App可能用到的知识点做一个归纳总结,让我们从一个简单的Hello world开始吧。

    Hello world

    # 步骤1:导入FastAPI
    from fastapi import FastAPI
    
    #步骤2:创建一个FastAPI“实例” 
    app = FastAPI()
    
    #步骤3:创建路径操作
    @app.get("/")
    async def root():
    #步骤4:定义路径操作功能
        return {"message": "Hello World"}
    
    
    if __name__ == '__main__':
        #步骤5:运行开发服务器
        import uvicorn
        uvicorn.run(app, host="127.0.0.1", port=8000)
    

    直接运行上面的代码或者命令行窗口运行uvicorn main:app --reload

    • main:文件main.py(Python“模块”)。
    • app:main.py在线内创建的对象app = FastAPI()。
    • --reload:更改代码后使服务器重新启动。仅用于开发。

    这样在浏览器输入http://127.0.0.1:8000/可以看到返回{"message":"Hello World"}

    下面让我们一点一点来扩展

    您可以在FastAPI应用程序app中配置几件事:title、description、version

    from fastapi import FastAPI
    
    app = FastAPI(
        title="My First Project",
        description="Let's have fun with it.",
        version="0.0.1",)
    
    @app.get("/")
    async def root():
        return {"message": "Hello World"}
    
    
    if __name__ == '__main__':
        import uvicorn
        uvicorn.run(app, host="127.0.0.1", port=8000)!
    

    FastAPI提供的自动API文档(http://127.0.0.1:8000/docs)会根据你的配置进行变更:

    [一]路径参数

    from fastapi import FastAPI
    
    app = FastAPI()
    
    
    @app.get("/users/me")
    async def read_user_me():
        return {"user_id": "the current user"}
    
    
    @app.get("/users/{user_id}")
    async def read_user(user_id: int):
        return {"user_id": user_id}
    

    注意路径操作是按顺序评估,因此如果你输入http://127.0.0.1:8000/users/me
    虽然以上两个路径参数都符合,但会按排在前面的进行执行。

    我们看看async def read_user(user_id: int) 这里的int是用来声明的,它有两个作用,数据转换和数据验证。

    如果你输入http://127.0.0.1:8000/users/3
    请注意,函数返回的值不是string "3", 而是int 3
    如果你输入http://127.0.0.1:8000/users/3.3,则会收到一个HTTP错误:

    {
        "detail": [
            {
                "loc": [
                    "path",
                    "user_id"
                ],
                "msg": "value is not a valid integer",
                "type": "type_error.integer"
            }
        ]
    }
    

    如果你要对一个路径进行操作, 可以这样:

    @app.get("/files/{file_path:path}")
    async def read_user_me(file_path: str):
        return {"file_path": file_path}
    

    :path告诉参数应与任何路径匹配,否则路径一般带有一个斜杠(/),可能不会被识别。例如/home/johndoe/myfile.txt, 注意在files和你的路径参数之间应该使用双斜杠(//),URL应该:http://127.0.0.1:8000/files//home/johndoe/myfile.txt

    路径“操作”其实是指HTTP“方法”之一,它是一个装饰器。

    常用的有

    • POST: 创建数据
    • GET: 读取数据
    • PUT:更新数据
    • DELETE:删除数据

    还有奇特一点的

    • OPTIONS
    • HEAD
    • PATCH
    • TRACE

    之前讲过数据验证是类型验证,进一步的,甚至可以进行数值验证。

    Path
    from fastapi import FastAPI, Path
    
    app = FastAPI()
    
    
    @app.get("/items/{item_id}")
    async def read_items(
        *,
        item_id: int = Path(..., title="The ID of the item to get", gt=0, le=1000),
        q: str,
    ):
        results = {"item_id": item_id}
        if q:
            results.update({"q": q})
        return results
    

    async 是异步关键字,异步的概念python很早就有,例如:

    @asyncio.coroutine
    def smart_fib(n):
    

    async / await 在python3.5+被引入,这里不详述, 对应上面的:

    async def  smart_fib(n):
    

    Path的第一个参数是默认值,由于路径参数始终需要一个参数,因为它必须是路径的一部分。这种情况下我们可以用...声明它,以将其标记为必需。实际上我们写任何默认值都不起作用。

    数值验证
    gt : greater than
    ge: greater than or equal
    lt : less than
    le : less than or equal


    我们可以将几个参数传递给路径操作装饰器以对其进行配置,请注意,这些参数直接传递给路径操作装饰器,而不是传递给路径操作函数。

    from typing import Set
    
    from fastapi import FastAPI, status
    from pydantic import BaseModel
    
    app = FastAPI()
    
    
    class Item(BaseModel):
        name: str
        description: str = None
        price: float
        tax: float = None
        tags: Set[str] = []
    
    
    @app.post("/items/", response_model=Item, 
                status_code=status.HTTP_201_CREATED,
                tags=["items"],
                summary="Create an item",
                description="Create an item with all",
                deprecated=False)
    async def create_item(*, item: Item):
        return item
    

    status_code(响应状态码): 也可以直接传递int代码,例如404
    tag(标签): 传递参数tags用list的str(通常只有一个str)
    summary(摘要)
    description(描述)
    deprecated(弃用):True代表在API文档它将被明确标记为不推荐使用,使用删除线

    [二]查询参数

    声明不属于路径参数的其他功能参数时,它们将自动解释为“查询”参数。

    from fastapi import FastAPI
    
    app = FastAPI()
    
    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,short: bool = False):
        return fake_items_db[skip : skip + limit]
    

    该查询是?URL中位于关键字之后的一组键值对,以&字符分隔。
    http://127.0.0.1:8000/items/?skip=0&limit=10

    我们还可以声明bool类型, 1 \true\on\yes 以及它们的大小写变体将被转换为True。
    http://127.0.0.1:8000/items/foo?short=yes

    当声明非路径参数的默认值时(目前,我们仅看到查询参数),则不需要此值。如果您不想添加特定值,而只是将其设为可选值,则将默认值设置为None。但是,当您需要一个查询参数时,就不能声明任何默认值

    例如下面这样一个例子:

    from fastapi import FastAPI
    
    app = FastAPI()
    
    
    @app.get("/items/{item_id}")
    async def read_user_item(item_id: str, needy: str, skip: int = 0, limit: int = None):
        item = {"item_id": item_id, "needy": needy, "skip": skip, "limit": limit}
        return item
    

    在这种情况下,有3个查询参数:

    needy,是必需的str。
    skip,int默认值为0。
    limit,可选的int。

    但是limitL int = None 会面临类型检查错误

    于是使用Optional(可选的类型声明),要改写成:

    from typing import Optional
    @app.get("/items/{item_id}")
    async def read_user_item(item_id: str, needy: str, skip: int = 0, limit: Optional[int] = None):
        item = {"item_id": item_id, "needy": needy, "skip": skip, "limit": limit}
        return item
    

    Query

    类似于路径参数使用Path来声明相同类型的验证, 可以使用Query为查询参数声明更多验证。

    from fastapi import FastAPI, Query
    
    app = FastAPI()
    
    
    @app.get("/items/")
    async def read_items(
        q: str = Query(None, min_length=3, max_length=50, regex="^fixedquery$")
    ):
        results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
        if q:
            results.update({"q": q})
        return results
    

    长度检查:min_length=3, max_length=50
    正则表达式检查:regex="^fixedquery$"

    假设您希望参数为item-query。像:http://127.0.0.1:8000/items/?item-query=foobaritems 但是item-query不是有效的Python变量名称。
    最接近的是item_query,但是您仍然需要它完全是item-query...

    然后,您可以声明一个alias,该别名将用于查找参数值

    from fastapi import FastAPI, Query
    
    app = FastAPI()
    
    
    @app.get("/items/")
    async def read_items(
        q: str = Query(
            None,
            alias="item-query",
            title="Query string",
            description="Query string for the items to search in the database that have a good match",
            min_length=3,
            max_length=50,
            regex="^fixedquery$",
            deprecated=True,
        )
    ):
        results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
        if q:
            results.update({"q": q})
        return results
    

    通用验证和元数据:

    • alias
    • title
    • description
    • deprecated

    特定于字符串的验证:

    • min_length
    • max_length
    • regex

    我觉得FastAPI一个非常nice的优势是它基于Pydantic模型来完成request body的许多校验工作,以避免在函数内部写很多类型处理,使代码看起来更简洁。下面一起看看。

    from fastapi import FastAPI
    from pydantic import BaseModel
    
    
    class Item(BaseModel):
        name: str
        description: str = None
        price: float
        tax: float = None
    
    
    app = FastAPI()
    
    
    @app.post("/items/")
    async def create_item(item: Item):
        return item
    

    item使用该类型声明,FastAPI将:

    • 以JSON读取请求的正文。
    • 转换相应的类型(如果需要)。
    • 验证数据。

    上面的该模型声明一个JSON“ object”(或Python dict),例如:

    {
        "name": "Foo",
        "description": "An optional description",
        "price": 45.2,
        "tax": 3.5
    }
    

    由于description和tax是可选的(默认值为None),此JSON“ object”也将有效:

    {
        "name": "Foo",
        "price": 45.2
    }
    

    复杂一点的,我们可以同时声明body,path和query参数:

    from fastapi import FastAPI
    from pydantic import BaseModel
    
    
    class Item(BaseModel):
        name: str
        description: str = None
        price: float
        tax: float = None
    
    
    app = FastAPI()
    
    
    @app.put("/items/{item_id}")
    async def create_item(item_id: int, item: Item, q: str = None):
        result = {"item_id": item_id, **item.dict()}
        if q:
            result.update({"q": q})
        return result
    

    功能参数将被识别如下:

    • 如果在path中也声明了该参数,它将用作路径参数。
    • 如果参数是一个的单一类型(如int,float,str,bool,等等)将被解释为一个查询参数。
    • 如果参数声明为Pydantic模型的类型,则它将被解释为请求正文。

    Body

    和的相同方法Query和Path为查询和路径参数定义额外的数据,FastAPI提供了等效的Body。

    from fastapi import Body, FastAPI
    from pydantic import BaseModel
    
    app = FastAPI()
    
    
    class Item(BaseModel):
        name: str
        description: str = None
        price: float
        tax: float = None
    
    
    class User(BaseModel):
        username: str
        full_name: str = None
    
    
    @app.put("/items/{item_id}")
    async def update_item(
        *, item_id: int, item: Item, user: User, importance: int = Body(...)
    ):
        results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
        return results
    

    如果这里的importance不使用Body,由于它是一个单一类型int, 它会被当做查询参数。但这里使用Body, 那么我们期望的json是下面这样子:

    {
        "item": {
            "name": "Foo",
            "description": "The pretender",
            "price": 42.0,
            "tax": 3.2
        },
        "user": {
            "username": "dave",
            "full_name": "Dave Grohl"
        },
        "importance": 5
    }
    

    注意这里多个body的时候(User、Item),会使用正文中的键作为参数名称,使得User和Item的属性值再往里嵌套一层。但如果是单个Body, 我们也想要使其属性值往里面嵌套一层的话,就要使用embed

    from fastapi import Body, FastAPI
    from pydantic import BaseModel
    
    app = FastAPI()
    
    
    class Item(BaseModel):
        name: str
        description: str = None
        price: float
        tax: float = None
    
    
    @app.put("/items/{item_id}")
    async def update_item(*, item_id: int, item: Item = Body(..., embed=True)):
        results = {"item_id": item_id, "item": item}
        return results
    

    这样,期望的body是:

    {
        "item": {
            "name": "Foo",
            "description": "The pretender",
            "price": 42.0,
            "tax": 3.2
        }
    }
    
    

    代替:

    {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2
    }
    

    如果对Item里的price 我们希望进行数据验证的话,也是有办法的,我们使用Field。

    from fastapi import Body, FastAPI
    from pydantic import BaseModel, Field
    
    app = FastAPI()
    
    
    class Item(BaseModel):
        name: str
        description: str = Field(None, title="The description of the item", max_length=300)
        price: float = Field(..., gt=0, description="The price must be greater than zero")
        tax: float = None
    
    
    @app.put("/items/{item_id}")
    async def update_item(*, item_id: int, item: Item = Body(..., embed=True)):
        results = {"item_id": item_id, "item": item}
        return results
    

    另外,example 是配置例子,例子会在docs API显示。

    from fastapi import Body, FastAPI
    from pydantic import BaseModel
    
    app = FastAPI()
    
    
    class Item(BaseModel):
        name: str
        description: str = None
        price: float
        tax: float = None
    
    
    @app.put("/items/{item_id}")
    async def update_item(
        *,
        item_id: int,
        item: Item = Body(
            ...,
            example={
                "name": "Foo",
                "description": "A very nice Item",
                "price": 35.4,
                "tax": 3.2,
            },
        )
    ):
        results = {"item_id": item_id, "item": item}
        return results
    

    我们甚至可以嵌套模型:

    from typing import Set
    
    from fastapi import FastAPI
    from pydantic import BaseModel, HttpUrl
    
    app = FastAPI()
    
    
    class Image(BaseModel):
        url: HttpUrl
        name: str
    
    
    class Item(BaseModel):
        name: str
        description: str = None
        price: float
        tax: float = None
        tags: Set[str] = []
        image: Image = None
    
    
    @app.put("/items/{item_id}")
    async def update_item(*, item_id: int, item: Item):
        results = {"item_id": item_id, "item": item}
        return results
    

    期望的body 是这样的:

    {
        "name": "Foo",
        "description": "The pretender",
        "price": 42.0,
        "tax": 3.2,
        "tags": ["rock", "metal", "bar"],
        "image": {
            "url": "http://example.com/baz.jpg",
            "name": "The Foo live"
        }
    }
    

    注意,在Image模型中,我们有一个url字段,我们可以将其声明为Pydantic的HttpUrl,而不是str, 该字符串将被检查为有效的URL,并在JSON Schema / OpenAPI中进行记录。


    下是一些您可以使用的其他数据类型:

    • UUID
      • 一个标准的“通用唯一标识符”,在许多数据库和系统中通常作为ID使用。
      • 在请求和响应中将以表示str
    • datetime.datetime
      • 一个Python datetime.datetime
      • 在请求和响应中,将以strISO 8601格式表示,例如:2008-09-15T15:53:00+05:00
    • datetime.date
      • Python datetime.date
      • 在请求和响应中,将以strISO 8601格式表示,例如:2008-09-15
    • datetime.time
      • 一个Python datetime.time
      • 在请求和响应中,将以strISO 8601格式表示,例如:14:23:55.003
    • datetime.timedelta
      • 一个Python datetime.timedelta
      • 在请求和响应中,将以float总秒数表示。
      • Pydantic还允许将其表示为“ ISO 8601时间差异编码”,有关更多信息请参阅文档
    • frozenset
      • 在请求和响应中,将与视为相同set
        • 在请求中,将读取列表,消除重复,并将其转换为set
        • 作为响应,set将会转换为list
        • 生成的架构将指定set值是唯一的(使用JSON架构的uniqueItems)。
    • bytes
      • 标准Python bytes
      • 在请求和响应中将被视为str
      • 生成的模式将指定,这是一个strbinary“格式”。
    • Decimal
      • 标准Python Decimal
      • 在请求和响应中,处理方式与相同float
    from datetime import datetime, time, timedelta
    from uuid import UUID
    
    from fastapi import Body, FastAPI
    
    app = FastAPI()
    
    
    @app.put("/items/{item_id}")
    async def read_items(
        item_id: UUID,
        start_datetime: datetime = Body(None),
        end_datetime: datetime = Body(None),
        repeat_at: time = Body(None),
        process_after: timedelta = Body(None),
    ):
        start_process = start_datetime + process_after
        duration = end_datetime - start_process
        return {
            "item_id": item_id,
            "start_datetime": start_datetime,
            "end_datetime": end_datetime,
            "repeat_at": repeat_at,
            "process_after": process_after,
            "start_process": start_process,
            "duration": duration,
        }
    

    相关文章

      网友评论

          本文标题:FastAPI for Fun

          本文链接:https://www.haomeiwen.com/subject/nqcsehtx.html