FastApi详细使用指南

简介

fastapi是高性能的web框架。他的主要特点是快速编码、减少人为bug、直观、简易、具有交互式文档、基于API的开放标准(并与之完全兼容):OpenAPI(以前称为Swagger)和JSON Schema

python版本

3.7 以上

安装

1
2
3
4
pip install fastapi  
pip install uvicorn
pip install pydantic
pip install sqlalchemy

fastapi: fastapi 框架包
uvicron: Uvicorn 是基于 uvloop 和 httptools 构建的非常快速的 ASGI 服务器
pydantic: 数据类型校验包
sqlalchemy: 数据库连接工具

项目结构梳理 (推荐)

1
2
3
4
5
6
7
8
9
10
11
12
13
├─ op_ns 
│ ├─ database 数据库资源
│ ├─ init.py
│ ├─ mysql.py 数据库连接配置
│ ├─ model 数据模型
│ ├─ router 路由层(控制层)
│ ├─ schemas 校验层
│ ├─ service 逻辑处理层
│ ├─ init.py
│ ├─ main.py 入口主文件
│ ├─ readme.md
│ ├─ dependencies.py 验签文件
│ ├─ requirements.txt 包文件

结构开发

mian.py 文件处理

op_ns/main.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
# -*- coding: utf-8 -*-
import uvicorn
from fastapi import Depends, FastAPI, Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from fastapi.encoders import jsonable_encoder

from op_ns.const import Code
from op_ns.dependencies import get_query_token, get_token_header
from op_ns.router import merchandise

app = FastAPI(
title="ns调用数据api",
description=
"""
fastapi mode
""",
version="0.1.0",
docs_url=None,
redoc_url=None,
)

# 注册路由
api_config = {
"prefix": "/api", # 前缀
"dependencies": [Depends(get_query_token), Depends(get_token_header)], # 进入api验签
"responses": {404: {"description": "Not found"}}
}

# 注册路由
app.include_router(apidoc.router)
app.include_router(merchandise.router, **api_config)


# 测试环境启动
if __name__ == '__main__':
uvicorn.run(app='main:app', host="0.0.0.0", port=8011, reload=True, debug=True)

验签处理

op_ns/dependencies.py

1
2
3
4
5
6
7
8
9
10
11
12
13
# -*- coding: utf-8 -*-
from fastapi import Header, HTTPException


# 这里的两个token放在header头里面
async def get_token_header(x_token: str = Header(...)):
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=400, detail="X-Token header invalid")


async def get_query_token(token: str = Header(...)):
if token != "hongju_oms_api":
raise HTTPException(status_code=400, detail="No Jessica token provided")

数据库连接

op_ns/database/mysql.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# -*- coding: utf-8 -*-
import traceback

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

from global_config.init_config import get_yaml_config

# 初始化数据库连接
config_dict = get_yaml_config()
engine = create_engine(config_dict['dsn']['erp_db'], echo=True)

# 创建对象的基类:
Base = declarative_base()

DBSession = sessionmaker(bind=engine)


def get_db():
db = DBSession()
try:
yield db
except Exception as e:
print(traceback.format_exc())
db.rollback()
finally:
db.close()

校验类

op_ns/schemas/merchandise.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
# -*- coding: utf-8 -*-
import time

from pydantic import BaseModel, constr
from typing import List, Optional


class MerchandiseInfoSchemas(BaseModel):
"""
商品数据验证类
"""
cat_class_name: constr(min_length=1) # 表示必填、字符长度不能为空
cat_name: constr(min_length=1)
category_parent_name: constr(min_length=1)
merchandise_sn: constr(min_length=1)
merchandise_name: constr(min_length=1)
base_merchandise_sn: constr(min_length=1)
is_customize: Optional[int] = 0 # 表示选填 类型为int,该字段可以不存在,默认值为0
is_group: Optional[int] = 0
merchandise_mode: int # 表示必填、类型为int
brand_name: constr(min_length=1)
is_embargo: Optional[int] = 0
merchandise_agent: Optional[str] = ''
manufacturer: Optional[str] = ''
is_validity_date: Optional[int] = 0
production_license_number: Optional[str] = ''
remark: Optional[str] = ''
is_disabled: Optional[int] = 0
is_deleted: Optional[int] = 0
created_at: Optional[int] = int(time.time())
updated_at: Optional[int] = int(time.time())

路由新增以及校验

op_ns/router/merchandise.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
# -*- coding: utf-8 -*-
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session

from op_ns.database.mysql import get_db
from op_ns.schemas.merchandise import MerchandiseInfoSchemas

router = APIRouter(
prefix="/merchandise", # 前缀
tags=["merchandise"] # tag
)


@router.post("/update", summary="商品资料更新")
async def merchandise_update(form_data: dict, db: Session = Depends(get_db)):
"""
接口描述
"""
data, err_msg = check_merchandise_update_data(form_data)
if not data:
return err_msg
process_res = {'code': Code.ERROR}
try:
merchandise_service = MerchandiseService(db)
process_res = merchandise_service.merchandise_update_process(data)
db.commit()
except Exception as e:
db.rollback()
process_res['msg'] = f'更新商品资料失败: %s' % e
finally:
return process_res


def check_merchandise_update_data(data: dict):
"""
商品资料新增|修改 数据校验
"""
try:
merchandise_info = data.get('merchandise_info')
specifications = data.get('specifications')
merchandise_suppliers = data.get('merchandise_suppliers')
merchandise_attributes = data.get('merchandise_attributes')
operation_type = data.get('operation_type')
if not operation_type:
raise Exception(f'operation_type is invalid value')
if not specifications:
raise Exception(f'specifications is invalid value')
if not merchandise_suppliers:
raise Exception(f'merchandise_suppliers is invalid value')

# 主表校验
merchandise_info_schemas = MerchandiseInfoSchemas(**merchandise_info) # 这里在使用校验类
# 规格校验
merchandise_specifications = []
for item in specifications:
merchandise_specifications.append(MerchandiseSpecificationsSchemas(**item))
# 供应商信息校验
merchandise_suppliers_list = []
for mer_supplier in merchandise_suppliers:
merchandise_suppliers_list.append(MerchandiseSuppliersSchemas(**mer_supplier))
# 属性校验
merchandise_attributes_list = []
if merchandise_attributes:
for mer_attr in merchandise_attributes:
merchandise_attributes_list.append(MerchandiseAttributesSchemas(**mer_attr))
return {
'merchandise_info': merchandise_info_schemas,
'merchandise_specifications': merchandise_specifications,
'merchandise_suppliers': merchandise_suppliers_list,
'merchandise_attributes': merchandise_attributes_list,
'operation_type': operation_type
}, {}
except Exception as e:
return None, {'code': Code.NO_PARAM, 'msg': "校验失败参数异常{e}".format(e=e)}

service以及数据库ORM

op/service/merchandise.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
# -*- coding: utf-8 -*-
import time

from op_ns.const import Code
from op_ns.model.merchandise import MerchandiseInfo, MerchandiseSpecifications, MerchandiseAlbum, \
UnauditedComboMerchandise, UnauditedComboMerchandiseSpecification, UnauditedComboMerchandiseDetails, \
MerchandiseSupplier, ProductAttribute, GroupedProductItem
from op_ns.database.mysql import DBSession


class MerchandiseService:

# 数据库初始化
def __init__(self, db: DBSession):
self.db = db

def merchandise_update_process(self, data):
"""
商品资料修改
"""
merchandise_info = data.get('merchandise_info')
merchandise_specifications = data.get('merchandise_specifications')
merchandise_suppliers = data.get('merchandise_suppliers', [])
merchandise_attributes = data.get('merchandise_attributes', [])
operation_type = data.get('operation_type')
merchandise_info_obj = self.db.query(MerchandiseInfo) \
.filter(MerchandiseInfo.merchandise_sn == merchandise_info.merchandise_sn) \
.filter(MerchandiseInfo.is_deleted == 0) \
.filter(MerchandiseInfo.is_disabled == 0)
# 判断新增 | 修改
# 更新操作
if operation_type == 'edit':
old_merchandise_info = merchandise_info_obj.first()
if not old_merchandise_info:
raise Exception(f"更新操作:获取货号%s信息失败" % merchandise_info.merchandise_sn)
# 主表更新
update_info_res = merchandise_info_obj.update(merchandise_info.dict())
if update_info_res != 1:
raise Exception(f'更新操作:更新merchandise_info信息失败')
merchandise_sn = old_merchandise_info.merchandise_sn
# 更新merchandise_specification
for item in merchandise_specifications:
spec_code = item.merchandise_specification_code
spec_data = dict()
spec_data['merchandise_id'] = old_merchandise_info.id
spec_data['merchandise_sn'] = merchandise_sn
spec_data['merchandise_specification_code'] = spec_code
spec_data['merchandise_specification_name'] = item.merchandise_specification_name
spec_data['base_id'] = item.base_id
spec_data['base_merchandise_sn'] = old_merchandise_info.base_merchandise_sn
spec_data['base_specification_code'] = item.base_specification_code
spec_data['barcode'] = item.barcode
spec_data['host_price'] = item.host_price
spec_data['measurement_unit'] = item.measurement_unit
spec_data['approval_number'] = item.approval_number
spec_data['exp_period'] = item.exp_period
spec_data['is_disabled'] = item.is_disabled
spec_data['is_deleted'] = item.is_deleted
spec_data['updated_at'] = int(time.time())
# 根据code查询数据
mer_spec_info_obj = self.db.query(MerchandiseSpecifications) \
.filter(MerchandiseSpecifications.merchandise_specification_code
== item.merchandise_specification_code) \
.filter(MerchandiseSpecifications.is_deleted == 0) \
.filter(MerchandiseSpecifications.is_disabled == 0)
have_spec = mer_spec_info_obj.first()
# 原数据有:修改;原数据没有:新增
if have_spec:
spec_id = have_spec.id
spec_data['created_at'] = have_spec.created_at
mer_spec_info_obj.update(spec_data)
# 没有:新增
else:
spec_data['created_at'] = int(time.time())
add_spec_data = MerchandiseSpecifications(**spec_data)
self.db.add(add_spec_data)
self.db.flush()
spec_id = add_spec_data.id
# 更新 merchandise_suppliers处理
for mer_supplier in merchandise_suppliers:
if mer_supplier.specification_code == spec_code:
mer_supplier_data = dict()
mer_supplier_data['merchandise_id'] = old_merchandise_info.id
mer_supplier_data['merchandise_sn'] = merchandise_sn
mer_supplier_data['specification_id'] = spec_id
mer_supplier_data['specification_code'] = spec_code
mer_supplier_data['base_id'] = mer_supplier.base_id
mer_supplier_data['base_merchandise_sn'] = mer_supplier.base_merchandise_sn
mer_supplier_data['base_specification_code'] = mer_supplier.base_specification_code
mer_supplier_data['supplier_name'] = mer_supplier.supplier_name
mer_supplier_data['is_main'] = mer_supplier.is_main
mer_supplier_data['before_tax_price'] = mer_supplier.before_tax_price
mer_supplier_data['tax_rate'] = mer_supplier.tax_rate
mer_supplier_data['cost_price'] = mer_supplier.cost_price
mer_supplier_data['is_free_tax'] = mer_supplier.is_free_tax
mer_supplier_data['other_price'] = mer_supplier.other_price
mer_supplier_data['taobao'] = mer_supplier.taobao
mer_supplier_data['guide_price'] = mer_supplier.guide_price
mer_supplier_data['retail_price'] = mer_supplier.retail_price
mer_supplier_data['is_deleted'] = mer_supplier.is_deleted
mer_supplier_data['updated_at'] = int(time.time())
# 根据sn和code查询
mer_supplier_obj = self.db.query(MerchandiseSupplier) \
.filter_by(merchandise_sn=old_merchandise_info.merchandise_sn,
specification_code=spec_code,
is_deleted=0)
have_mer_supplier = mer_supplier_obj.first()
# 原数据有:修改;原数据没有:新增
if have_mer_supplier:
mer_supplier_data['created_at'] = have_mer_supplier.created_at
mer_supplier_obj.update(mer_supplier_data)
# 没有:新增
else:
mer_supplier_data['created_at'] = int(time.time())
add_mer_supper_data = MerchandiseSupplier(**mer_supplier_data)
self.db.add(add_mer_supper_data)
self.db.flush()
# 更新 product_attributes处理
for pro_attr in merchandise_attributes:
if spec_code == pro_attr.merchandise_specification_code:
check_data = {
"merchandise_id": old_merchandise_info.id,
"merchandise_sn": merchandise_sn,
"specification_code": spec_code,
"specification_id": spec_id
}
# 添加|修改 执行
self.create_and_edit_product_attributes(pro_attr.dict(), check_data)
return {
'code': Code.SUCCESS,
'msg': f'更新商品资料成功'
}
# 新增操作
elif operation_type == 'create':
old_merchandise_info = merchandise_info_obj.first()
if old_merchandise_info:
raise Exception(f"创建操作:已存在货号%s信息,无法再次新增" % merchandise_info.merchandise_sn)
merchandise_info_obj = MerchandiseInfo(**merchandise_info.dict())
self.db.add(merchandise_info_obj)
self.db.flush()
merchandise_info_id = merchandise_info_obj.id
if not merchandise_info_id:
raise Exception(f'创建动作:新增merchandise_info失败')
add_mer_specs_obj = []
for item in merchandise_specifications:
specs_items = MerchandiseSpecifications(**item.dict())
specs_items.merchandise_id = merchandise_info_id
specs_items.merchandise_sn = merchandise_info.merchandise_sn
specs_items.base_merchandise_sn = merchandise_info.base_merchandise_sn
add_mer_specs_obj.append(specs_items)
# merchandise_specifications 处理
for item in add_mer_specs_obj:
self.db.add(item)
self.db.flush()
new_spec_id = item.id
# merchandise_suppliers 处理
for mer_supplier in merchandise_suppliers:
if item.merchandise_specification_code == mer_supplier.specification_code:
merchandise_supplier_obj = MerchandiseSupplier(**mer_supplier.dict())
merchandise_supplier_obj.merchandise_id = merchandise_info_id
merchandise_supplier_obj.merchandise_sn = merchandise_info.merchandise_sn
merchandise_supplier_obj.specification_id = new_spec_id
self.db.add(merchandise_supplier_obj)
self.db.flush()
# merchandise_attributes 处理
if merchandise_attributes:
for mer_attr in merchandise_attributes:
if item.merchandise_specification_code == mer_attr.merchandise_specification_code:
add_attr_data = {
"merchandise_id": merchandise_info_id,
"merchandise_sn": merchandise_info.merchandise_sn,
"specification_id": new_spec_id,
"specification_code": item.merchandise_specification_code
}
# 执行添加
self.create_and_edit_product_attributes(mer_attr.dict(), add_attr_data)

return {
'code': Code.SUCCESS,
'msg': f'新增商品资料成功'
}
else:
return {
"code": Code.ERROR,
"msg": f"接口operation_type 错误"
}

启动

端口 8011
直接执行 main.py 文件

接口文档

文档会自动生成

文档地址:
http://0.0.0.0:8011/docs

文档加密处理

op_ns/router/apidoc.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# -*- coding: utf-8 -*-
"""
api 文档配置 默认关闭api文档,需要通过密码设置才能正常使用
"""
from fastapi import APIRouter, Depends, HTTPException
from fastapi.openapi.docs import get_swagger_ui_html
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from starlette.status import HTTP_401_UNAUTHORIZED

router = APIRouter()
security = HTTPBasic()


@router.get("/docs", include_in_schema=False)
@router.get("/redoc", include_in_schema=False)
async def get_documentation(credentials: HTTPBasicCredentials = Depends(security)):
if credentials.username != "root" or credentials.password != "4371c4ad1f64194c3564c55eb8090f58":
raise HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="账号或密码无效",
headers={"WWW-Authenticate": "Basic"},
)

else:
return get_swagger_ui_html(openapi_url="/openapi.json", title="docs")
谢翔 wechat
坚持原创技术分享,您的支持将鼓励我继续创作!
-------------本文结束感谢您的阅读-------------
0%