代码框架

• 整个项目包含api层、业务逻辑层和service层，在api层进行入参格式校验，业务层处理具体的业务信息，在service层进行返回格式校验，使用的框架为FastApi

Api层

基于pydantic设置参数的类型，对入参进行格式校验

• 入参校验示例

from typing import Optional
from pydantic import BaseModel, ValidationError, validator
from datetime import datetime

class DatetimeMixin(BaseModel):
    startDatetime: Optional[datetime] = '2000-01-01T00:00:00' # 设定时间的默认值
    endDatetime: Optional[datetime] = '2100-01-01T00:00:00'

class PageInfo(BaseModel):
    page: int = 1  # 设定参数为int类型
    size: int = 10

    @validator('page', 'size')
    def check_page(cls, v):  # 检验字段名是否符合要求
        if v > 0:
            return v
        else:
            raise ValidationError

• 出参校验示例

from pydantic import BaseModel

class ResponseModel(BaseModel):  # 设置状态码和信息，直接继承
    statusCode: int
    message: str

接口实现

fastapi使用 '.' 操作符获取参数，应该在这一层获取所有具体参数，传给service

• get实现,入参使用Depends注入依赖，表示将类对象RarioParams传递给query

from fastapi import APIRouter, BackgroundTasks, UploadFile, File, Depends
router = APIRouter()

@router.get('/ratio', response_model=RatioResponse, tags=Tag.Analysis.value)
def ratio(query: RatioParams = Depends()):
    key = query.key
    start = query.startDatetime
    end = query.endDatetime
    response = ratio_service.ratio_analysis(key, start, end)
    return response

• post实现

@router.post('/aggregation', response_model=AggregationResponse, tags=Tag.Analysis.value)  # tag是swagger的显示栏标签
def aggregation(query: AggregationQuery):
    aggs = query.agg
    filters = query.filter
    page_info = query.pageInfo
    order = query.order
    aggregation_service = AggregationService(aggs, page_info, filters, order)
    response = aggregation_service.aggregation_analysis()

    return response

• 后台任务

@router.post('/upload', response_model=ResponseModel, tags=Tag.DataManagement.value)
def upload(background_tasks: BackgroundTasks, file: UploadFile = File(...)):  # ...Ellipsis对象，相当于占位符
    background_tasks.add_task(service.upload_task, file.file, file.filename)
    return service.response()

写着写着发现知识太零碎没法写了，还是过几天专门记录一下fastapi的学习吧