Fastapi框架-冷饭再炒-基础知识补充篇(2)

上一篇中说过,希望自己能坚持继续整理自己的在学习python web开发框架的Fastapi的相关的笔记。既然有了开头,所以也希望自己有个结尾,即使没结尾~也至少还有一个过程吧!所以继续整理之前我再使用过程遇到一些问题!

Fastapi 基础补充扩展

上一篇中相关的知识的介绍主要是官网的提供的示例多,仅仅是自己基于上面进行操作实际演练,而实际应用上,涉及的问题还是比较多,对于基础的上,其实还有很多需要注意的点。

1 Fastapi app对象初始化的配置项说明

app = FastAPI()
复制代码

从实例化一个应用开始,我们的FastAPI()就只是使用默认的参数,但是里面涉及的配置项还是比较多。这里我补充一下。

从配置项里我们可以学到主要有:

  • 1:如何对app对象进行相关的描述
  • 2:如何的配置我们的api文档的地址和关闭这个文档地址
  • 3:如何添加APP启动和关闭的事件回调的通知

首先我们的进入内部源码看到的是:

class FastAPI(Starlette):
    def __init__(
        self,
        *,
        debug: bool = False,
        routes: Optional[List[BaseRoute]] = None,
        title: str = "FastAPI",
        description: str = "",
        version: str = "0.1.0",
        openapi_url: Optional[str] = "/openapi.json",
        openapi_tags: Optional[List[Dict[str, Any]]] = None,
        servers: Optional[List[Dict[str, Union[str, Any]]]] = None,
        dependencies: Optional[Sequence[Depends]] = None,
        default_response_class: Type[Response] = Default(JSONResponse),
        docs_url: Optional[str] = "/docs",
        redoc_url: Optional[str] = "/redoc",
        swagger_ui_oauth2_redirect_url: Optional[str] = "/docs/oauth2-redirect",
        swagger_ui_init_oauth: Optional[Dict[str, Any]] = None,
        middleware: Optional[Sequence[Middleware]] = None,
        exception_handlers: Optional[
            Dict[
                Union[int, Type[Exception]],
                Callable[[Request, Any], Coroutine[Any, Any, Response]],
            ]
        ] = None,
        on_startup: Optional[Sequence[Callable[[], Any]]] = None,
        on_shutdown: Optional[Sequence[Callable[[], Any]]] = None,
        openapi_prefix: str = "",
        root_path: str = "",
        root_path_in_servers: bool = True,
        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
        callbacks: Optional[List[BaseRoute]] = None,
        deprecated: Optional[bool] = None,
        include_in_schema: bool = True,
        **extra: Any,
    ) -> None:
    `````
    `````
复制代码

里面涉及的参数还真不是一一般多的吧,但是所有的项都是非必传的项,都有相关的默认值。

这里我就简单给一些参数做简单的注释:

  • debug: bool = False 是否开启debug

  • routes: Optional[List[BaseRoute]] = None app 对象下挂在的所有的路由列表

  • title: str = “FastAPI”, 给我们app应用的文档定义一个标题文,会对应到我们的文档上的标题

  • description: str = “” app应用的文档具体描述

  • version: str = “0.1.0”, app应用的文档版本号信息

image.png

  • openapi_prefix: str = “” 访问的api地址的访问的前缀

  • openapi_url: Optional[str] = “/openapi.json” api文档描述的整个JSON描述访问的URL地址,

PS:后续可以直接导入这个到 swagger,就可以看到我们的应用的所有的接口。【Swagger对于的openapi规范的文档格式】

如使如下的示例展示我们的API文档:

 return get_swagger_ui_html(
        openapi_url=openapi_url,
        title="Swagger UI",
        oauth2_redirect_url=oauth2_redirect_url,
        init_oauth=app.swagger_ui_init_oauth,
        swagger_js_url="https://cdn.jsdelivr.net/npm/swagger-ui-dist@3.48.0/swagger-ui-bundle.js",
        swagger_css_url="https://cdn.jsdelivr.net/npm/swagger-ui-dist@3.48.0/swagger-ui.css",
    )
    )
复制代码

image.png

  • openapi_tags: Optional[List[Dict[str, Any]]] = None 每个API文档接口的标签

image.png

  • servers: Optional[List[Dict[str, Union[str, Any]]]] = None 暂时不知道可以用来做什么

  • dependencies: Optional[Sequence[Depends]] = None 全局依赖的对象(一般起全局作用)

  • default_response_class: Type[Response] = Default(JSONResponse) 默认响应报文返回对象,它返回application/json格式的response。这个是FastAPI默认返回的response。

  • docs_url: Optional[str] = “/docs” swagger api文档的访问的地址,可以自定义,

PS: 也可以去除设置为None则就是不存在这个API文档

  • redoc_url: Optional[str] = “/redoc”, redocapi文档的访问的地址,可以自定义,

  • swagger_ui_init_oauth: Optional[Dict[str, Any]] = None, swagger ui api文档 授权初始化

  • swagger_ui_oauth2_redirect_url: Optional[str] = “/docs/oauth2-redirect” swagger api文档的访问的授权的回调地址

  • middleware: Optional[Sequence[Middleware]] = None 应用的中间件

  • exception_handlers: Optional[
    Dict[
    Union[int, Type[Exception]],
    Callable[[Request, Any], Coroutine[Any, Any, Response]],
    ]
    ] = None 应用的异常错误处理器对象,可以传入对于的函数进行的对应异常的处理

示例如:

from starlette.responses import Response


async def Not404Fount(request: Request, exc):
   response = Response('{"Error": "500"}')
   return response


async def Not500Fount(request: Request, exc):
   msg = ''
   if isinstance(exc, ZeroDivisionError):
       msg = 'ZeroDivisionError'
   response = Response(msg)
   return response


# 添加404
app.add_exception_handler(404, Not404Fount)
# 添加404
app.add_exception_handler(500, Not500Fount)

复制代码
  • on_startup: Optional[Sequence[Callable[[], Any]]] = None 应用启动的时候的回调函数

  • on_shutdown: Optional[Sequence[Callable[[], Any]]] = None 应用退出的时候的回调函数

启动和关闭的回调的示例:

def f_startup():
    print("启动起来")
    pass


def f_shutdown():
    print("关闭")
    pass


# 启动回调
app.add_event_handler('startup', f_startup)
# 关闭回调
app.add_event_handler('shutdown', f_shutdown)
复制代码
  • root_path: str = “”, 当前应用所处的根目录

  • root_path_in_servers: bool = True,

  • deprecated: Optional[bool] = None 标记当前的接口的文档当前版本是否可用,设置为True的情况下:

image.png

  • include_in_schema: bool = True, 是否开启API文档的查看,False的话所有的从生成的OpenAPI架构(以及自动文档系统)会自动清除

image.png

2 配置项参数deprecated和include_in_schema其他说明

在app对象里面我们的deprecated和include_in_schema是起到全局的作用的,但是如果仅仅只是针对某些接口的进行相关接口的排除在文档中,或是进行标记当前版本的可用性的话,如果操作呢?

示例代码:

import uvicorn

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

from fastapi import BackgroundTasks, FastAPI, Depends

app = FastAPI(
    title="XXXX消息管理系统",
    description='这个系统主要是用于XXXXXX',
    version='v1.0.0'
)

import threading

@app.get("/sync3")
def sync3():
    print('sync-当前线程的名称:', threading.current_thread().getName())

    return "sync任务已经处理完成!"


# 隐藏这个接口在Api文档中不在显现
@app.get("/sync",include_in_schema=False)
def sync():
    print('sync-当前线程的名称:', threading.current_thread().getName())

    return "sync任务已经处理完成!"

# 这个接口在Api文档中 标记为已经无法使用,相当于过期,不建议再使用
@app.get("/async",deprecated=True)
async def asyncaction():
    print(app.routes)
    print('asyncaction-当前线程的名称:', threading.current_thread().getName())
    return "async任务已经处理完成!"


if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务:uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

复制代码

文档展示结果:

image.png

3 写好API接口文档描述

PS:要想看到文档,第一次肯定是需要网络滴哟!

3.1 接口基本的文档描述

我们知道这个swagger API 文档是挺好用,但是一些接口的API的描述,我们需要配置上,不然我们的也不知道这个接口干嘛!其实说白了,就是我是英文的菜鸟,需要中文来解读!哈哈

比如这个接口,完全不知道用来干嘛!
image.png

一个好的API接口文档还是有需要滴!

示例如:

#!/usr/bin/evn python
# coding=utf-8


import uvicorn

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

from fastapi import BackgroundTasks, FastAPI, Depends

app = FastAPI(
    title="XXXX消息管理系统",
    description='这个系统主要是用于XXXXXX',
    version='v1.0.0'
)

import threading


@app.get("/sync3")
def sync3():
    print('sync-当前线程的名称:', threading.current_thread().getName())

    return "sync任务已经处理完成!"


# 隐藏这个接口在Api文档中不在显现
@app.get("/sync", include_in_schema=False)
def sync():
    print('sync-当前线程的名称:', threading.current_thread().getName())

    return "sync任务已经处理完成!"


# 这个接口在Api文档中 标记为已经无法使用,相当于过期,不建议再使用
@app.get("/async", deprecated=True)
async def asyncaction():
    print(app.routes)
    print('asyncaction-当前线程的名称:', threading.current_thread().getName())
    return "async任务已经处理完成!"


@app.get(path='/api/v1/get/user', summary='获取用户', description='这个接口是用来添加用户的', tags=['用户模块'])
def getuser():
    return {"code": 0, "msg": "获取成功", "data": None}


@app.post(path='/api/v1/add/user', summary='添加用户', description='这个接口是用来获取用户信息的', tags=['用户模块'])
def adduser():
    return {"code": 0, "msg": "添加成功", "data": None}


@app.put(path='/api/v1/updata/user', summary='更新用户', description='这个接口是用来更新用户信息的', tags=['用户模块'])
def updatauser():
    return {"code": 0, "msg": "修改成功", "data": None}


@app.put(path='/api/v1/delete/user', summary='删除用户', description='这个接口是用来删除用户信息的', tags=['用户模块'])
def deleteuser():
    return {"code": 0, "msg": "删除成功", "data": None}


@app.put(path='/api/v1/add/data', summary='新增数据', description='这个接口是用来新增数据', tags=['数据模块'])
def adddatas():
    return {"code": 0, "msg": "删除成功", "data": None}


if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务:uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

复制代码

文档效果示例:

image.png

image.png

3.2 接口相关参数文档描述

查看我们的源码

Query:源码

def Query(  # noqa: N802
    default: Any,
    *,
    alias: Optional[str] = None,
    title: Optional[str] = None,
    description: Optional[str] = None,
    gt: Optional[float] = None,
    ge: Optional[float] = None,
    lt: Optional[float] = None,
    le: Optional[float] = None,
    min_length: Optional[int] = None,
    max_length: Optional[int] = None,
    regex: Optional[str] = None,
    example: Any = Undefined,
    examples: Optional[Dict[str, Any]] = None,
    deprecated: Optional[bool] = None,
    **extra: Any,
    
    
基于Query模块声明缺省值

- 可选参数声明 
- 形式:q: str = Query(None)  # 等同于  q: str = None


- 默认值参数声明
- 形式: q: str = Query("query")  # 等同于  q: str = "query"

- 必选参数声明
- 形式: q: str = Query(...)  # 等同于 q: str

自动化文档的附加信息主要参数:

- title:参数标题
- description:参数描述信息
- deprecated:表示参数即将过期,或不可用
特殊附加信息:
- alias:参数别名
    
复制代码

Body 源码:

def Body(  # noqa: N802
    default: Any,
    *,
    embed: bool = False,
    media_type: str = "application/json",
    alias: Optional[str] = None,
    title: Optional[str] = None,
    description: Optional[str] = None,
    gt: Optional[float] = None,
    ge: Optional[float] = None,
    lt: Optional[float] = None,
    le: Optional[float] = None,
    min_length: Optional[int] = None,
    max_length: Optional[int] = None,
    regex: Optional[str] = None,
    example: Any = Undefined,
    examples: Optional[Dict[str, Any]] = None,
    **extra: Any,
) -> Any:


自动化文档的附加信息主要参数:

embed=True 加上这个会我们的示例example就不会显示出来
- title:参数标题
- description:参数描述信息
- deprecated:表示参数即将过期,或不可用
特殊附加信息:
- alias:参数别名
example : 示例
examples:示例列表
复制代码

他们的参数上都代理对于的相关的参数作用描述还有相关的示例传入。

所以我们的可以用上面的展示我们的参数的说明和解释.

示例如:


import uvicorn

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

from fastapi import BackgroundTasks, FastAPI, Depends, Body, Query
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI(
    title="XXXX消息管理系统",
    description='这个系统主要是用于XXXXXX',
    version='v1.0.0'
)

import threading


@app.get("/sync3")
def sync3():
    print('sync-当前线程的名称:', threading.current_thread().getName())

    return "sync任务已经处理完成!"


# 隐藏这个接口在Api文档中不在显现
@app.get("/sync", include_in_schema=False)
def sync():
    print('sync-当前线程的名称:', threading.current_thread().getName())

    return "sync任务已经处理完成!"


# 这个接口在Api文档中 标记为已经无法使用,相当于过期,不建议再使用
@app.get("/async", deprecated=True)
async def asyncaction():
    print(app.routes)
    print('asyncaction-当前线程的名称:', threading.current_thread().getName())
    return "async任务已经处理完成!"


@app.get(path='/api/v1/get/user', summary='获取用户', description='这个接口是用来添加用户的', tags=['用户模块'])
def getuser():
    return {"code": 0, "msg": "获取成功", "data": None}


@app.post(path='/api/v1/add/user', summary='添加用户', description='这个接口是用来获取用户信息的', tags=['用户模块'])
def adduser():
    return {"code": 0, "msg": "添加成功", "data": None}


@app.put(path='/api/v1/updata/user', summary='更新用户', description='这个接口是用来更新用户信息的', tags=['用户模块'])
def updatauser():
    return {"code": 0, "msg": "修改成功", "data": None}


@app.delete(path='/api/v1/delete/user', summary='删除用户', description='这个接口是用来删除用户信息的', tags=['用户模块'])
def deleteuser():
    return {"code": 0, "msg": "删除成功", "data": None}


@app.put(path='/api/v1/add/data', summary='新增数据', description='这个接口是用来新增数据', tags=['数据模块'])
def adddatas():
    return {"code": 0, "msg": "新增数据", "data": None}


@app.get("/api/v1/ceshi/data", summary='查看数据', description='这个接口是用来新增数据', tags=['数据模块'])
async def read_items(
        uerid: str = Query(..., min_length=3, max_length=50,
                           title='用户uerid',
                           description='这个必须,不然找不到用户滴',
                           alias='UID',
                           example="UUU323423385783785"),
        uername: str = Query(..., min_length=3, max_length=50,
                             title='用户uername',
                             description='用户昵称',
                             alias='UNAME',
                             example="小钟同学"),

):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    print(uername)
    if uerid:
        results.update({"uerid": uerid})
    return results


class Item(BaseModel):
    name: str
    description: str = Field(None, title="描述", description="描写的详细内容", max_length=300)
    price: float = Field(..., gt=0, title="价格", description="价格具体说明")
    tax: float = None

# class Item(BaseModel):
#     name: str
#     description: str = None
#     price: float
#     tax: float = None

@app.post("/api/v1/udaoay/data", summary='修改数据', description='这个接口是用来新增数据', tags=['数据模块'])
async def uddata(item: Item = Body(..., embed=False, example={'para1': "必填,格式要求是字符串", 'name': "必填,格式要求是字符串",
                                                             'description': "选填,格式要求是字符串,默认值是None",
                                                             'price': "必填,格式要求是float",
                                                             'tax': "选填,格式要求是浮点型"})):
    return {"code": 0, "msg": "修改成功", "data": item}



if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务:uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

复制代码

那关于查询参数的结果示例描述如:

image.png

关于请求体的参数的结果示例描述如:

image.png

image.png

3.3 使用模型的Config来描述接口参数的示例

class Item(BaseModel):
    name: str
    description: str = Field(None, title="描述", description="描写的详细内容", max_length=300)
    price: float = Field(..., gt=0, title="价格", description="价格具体说明")
    tax: float = None

    class Config:

        schema_extra={
            "example":
                {
                    "name":"描写的详细内容",
                    "price": "价格具体说明",
                }
        }


@app.post("/api/v1/udaoay/data", summary='修改数据', description='这个接口是用来新增数据', tags=['数据模块'])
async def uddata(item: Item = Body(..., embed=False, example={'para1': "必填,格式要求是字符串", 'name': "必填,格式要求是字符串",
                                                             'description': "选填,格式要求是字符串,默认值是None",
                                                             'price': "必填,格式要求是float",
                                                             'tax': "选填,格式要求是浮点型"})):
    return {"code": 0, "msg": "修改成功", "data": item}
复制代码

image.png

4 API安全性问题

API文档的对于我们的项目的而言就是所有的入口,安全性势必也是不可忽视的。对于安全性的方案大致有:

  • 1:线上环境关闭API接口的文档,取消API接口的文档的地址
  • 2:全局性的关闭API文档的所有接口,通过全局的include_in_schema来关闭
  • 3:仅仅支持内网可见的API部署一个实例出来
  • 4: 通过Nginx 启用 Base Auth 实现用户认证或对应的IP访问限制某API地址
  • 5:APP应用上对请求来源IP做限制
  • 6:App应用上也启用用户认知机制

补充App内自己用户和密码认知的一种简单的方式:

import uvicorn

from fastapi import Depends
from fastapi import FastAPI, File, UploadFile
from config import settings

from fastapi.security import HTTPBasic, HTTPBasicCredentials
from fastapi.exceptions import HTTPException
from fastapi.openapi.docs import get_swagger_ui_html
from starlette.status import HTTP_401_UNAUTHORIZED

app = FastAPI(
    title="XXXX消息管理系统",
    description='这个系统主要是于XXXXXX',
    version=settings.API_V1_STR,
    # 自定义开启的API文档的接口的时候,需要设置默认的关闭
    docs_url=None
)


security = HTTPBasic()


@app.get("/docs",include_in_schema=False)
async def get_documentation(credentials: HTTPBasicCredentials = Depends(security)):
    print("sssssssssssssssssss")
    if credentials.username != "xiaozhongtongxue" or credentials.password != "xiaozhongtongxue":
        raise HTTPException(
            status_code=HTTP_401_UNAUTHORIZED,
            detail="y用户和密码不对!!!!!",
            headers={"WWW-Authenticate": "Basic"},
        )
    else:
        return get_swagger_ui_html(openapi_url="/openapi.json", title="doc")


if __name__ == '__main__':
    # 等于通过 uvicorn 命令行 uvicorn 脚本名:app对象 启动服务:uvicorn xxx:app --reload
    uvicorn.run('frun:app', host="127.0.0.1", port=8000, debug=True, reload=True)

复制代码

image.png

PS:需要开启APP内的认证的话,需要设置默认的关闭docs_url=None

5 API响应报文类型

对于响应报文,通常我们的写API接口的肯定是一般返回json格式的报文体,但是也不排除我单纯只是需要返回字符串,或者返回HTML模板,返回的是数据流,返回XML格式等。

所以这里也需要有所针对性的进行定制,但是我们的Fastapi其实基本上都封装好了,不需要我们再自己去处理了.

PS如果存在自定义的错误参数的校验的时候,直接的返回字符串类似的是会报错的,必须使用PlainTextResponse才能正确返回。

如下示例:

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    # return JSONResponse({'mes':'触发了RequestValidationError错误,,错误信息:%s 你妹的错了!'%(str(exc))})
   # TypeError: 'str' object is not callable 
    return "小姐姐在哪里!"

需要返回使用 return PlainTextResponse("小姐姐在哪里!")才正确
复制代码

5.1responses 类型

from fastapi.responses import (HTMLResponse, JSONResponse, ORJSONResponse, Response, PlainTextResponse, UJSONResponse,StreamingResponse,FileResponse)
复制代码
  • Response 需要自己的再扩展的基础响应体
  • HTMLResponse 返回HTML模板
  • JSONResponse jison格式
  • ORJSONResponse 使用了另一个库的JSON格式
  • UJSONResponse 使用了另一个库的JSON格式
  • StreamingResponse 输出流的格式
  • FileResponse 文件响应体的方式

5.2 自定义Response

示例如返回XML格式响应:

from fastapi import FastAPI, Response

app = FastAPI()


@app.get("/legacy/")
def get_legacy_data():
    data = """<?xml version="1.0"?>
    <shampoo>
    <Header>
        Apply shampoo here.
    </Header>
    <Body>
        You'll have to use soap here.
    </Body>
    </shampoo>
    """
    return Response(content=data, media_type="application/xml")
复制代码

5.3 返回字符串PlainTextResponse

示例如:



@app.get("/PlainTextResponse1/")
async def PlainTextResponse1():
    return PlainTextResponse('你好啊')


@app.get("/PlainTextResponse2/", response_class=PlainTextResponse)
async def PlainTextResponse2():
    return "你好!"

复制代码

5.4 返回JSON数据格式

示例如:

@app.post("/api/v1/udaoay/data")
async def uddata():
    return {"code": 0, "msg": "修改成功", "data": None}
    或
    return JSONResponse(status_code=404, content={"message": "Item not found"})

复制代码

5.5 返回HTMLResponse

示例如:

from fastapi import FastAPI
from fastapi.responses import HTMLResponse

app = FastAPI()


def generate_html_response():
    html_content = """
    <html>
        <head>
            <title>Some HTML in here</title>
        </head>
        <body>
            <h1>Look ma! HTML!</h1>
        </body>
    </html>
    """
    return HTMLResponse(content=html_content, status_code=200)


@app.get("/items/", response_class=HTMLResponse)
async def read_items():
    return generate_html_response()
复制代码

5.6 返回RedirectResponse临时重定向

示例如:


@app.get("/RedirectResponse/", response_class=HTMLResponse)
async def RedirectResponse():
   return RedirectResponse("https://wwww.baifuw.com")
复制代码

5.7 返回流信息StreamingResponse

示例1如:
接受异步生成器或普通生成器/迭代器,并对响应体进行流。


from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()


async def fake_video_streamer():
    for i in range(10):
        yield b"some fake video bytes"


@app.get("/")
async def main():
    return StreamingResponse(fake_video_streamer()
复制代码

示例2如:
如果您有一个类似文件的对象(例如,由open()),您可以在StreamingResponse.

这包括许多与云存储、视频处理等交互的库。


from fastapi import FastAPI
from fastapi.responses import StreamingResponse

some_file_path = "large-video-file.mp4"
app = FastAPI()


@app.get("/")
def main():
    file_like = open(some_file_path, mode="rb")
    return StreamingResponse(file_like, media_type="video/mp4")
复制代码

5.8 返回FileResponse

异步地将文件作为响应流。

使用与其他响应类型不同的参数集实例化:

  • path-要流的文件的文件路径。
  • headers-任何要包括的自定义标头,作为字典。
  • media_type-显示媒体类型的字符串。如果未设置,文件名或路径将用于推断媒体类型。
  • filename-如果已设定,这将包括在答复中Content-Disposition.

文件响应将包括适当的Content-Length, Last-Modified和ETag标题。

from fastapi import FastAPI
from fastapi.responses import FileResponse

some_file_path = "large-video-file.mp4"
app = FastAPI()


@app.get("/")
async def main():
    return FileResponse(some_file_path)
复制代码

6 BaseSettings 配置文件读取

以往写Flask的时候,通常我们的会把一些配置信息,写到一个配置的文件里面,然后是根据测试环境还是正式环境的的方式来区分读取相关的配置值。

定义位置文件Congfig.py

from functools import lru_cache
import secrets
from pydantic import AnyHttpUrl, BaseSettings

import pprint

pp = pprint.PrettyPrinter(indent=4)


class Settings(BaseSettings):
    API_V1_STR: str = "/api/v1"
    SECRET_KEY: str = secrets.token_urlsafe(32)

    JWT_SECRET_KEY: str = ''
    JWT_ALGORITHM: str = ''
    JWT_EXPIRES: int = 60
    JWT_REFRESH_EXPIRES_DAYS: int = 1

    class Config:
        env_file = ".env"
        case_sensitive = True
        env_file_encoding = 'utf-8'


@lru_cache()
def get_settings():
    return Settings()


print("config.py > settings : ...")
settings = Settings()
pp.pprint(settings.dict())

复制代码

运行输出:

config.py > settings : ...
{   'API_V1_STR': '/api/v1',
    'JWT_ALGORITHM': '',
    'JWT_EXPIRES': 60,
    'JWT_REFRESH_EXPIRES_DAYS': 1,
    'JWT_SECRET_KEY': '',
    'SECRET_KEY': 'pGR3SgBuuMq-nbmjRTJHh-px4huaPpTpgQwnM5MDcik'}
复制代码

然后在我们App初始化的时候引入我们上面的单列对象settings


import uvicorn

from fastapi import BackgroundTasks, FastAPI, Depends, Body, Query
from fastapi.responses import JSONResponse,ORJSONResponse,UJSONResponse,PlainTextResponse,HTMLResponse,StreamingResponse,RedirectResponse,FileResponse,Response
from fastapi.exceptions import RequestValidationError
from fastapi import FastAPI, File, UploadFile
from config import settings

app = FastAPI()


app = FastAPI(
    title="XXXX消息管理系统",
    description='这个系统主要是于XXXXXX',
    version=settings.API_V1_STR
)

复制代码

总结

这是基础的只是第二篇,纯属个人的学习笔记,如有纰漏还希望多多指点。

后续打算针对安全和数据库这些都做专题的梳理。下一期再见!

END

小钟同学 | 文 【原创】【转载请联系本人】| QQ:308711822

© 版权声明
THE END
喜欢就支持一下吧
点赞0 分享