mirror of https://gitee.com/anolis/sysom.git
218 lines
7.8 KiB
Markdown
218 lines
7.8 KiB
Markdown
# SysOM 框架 SDK
|
||
|
||
## 1. 框架基础能力
|
||
|
||
## 2. `fastapi_helper`
|
||
|
||
`fastapi_helper` 对使用 fastapi 编写微服务的一些常用操作进行了封装，可以提供如下开箱即用的能力：
|
||
- 查询辅助（支持分页、排序、按字段过滤等等）
|
||
- 响应辅助（支持快速返回标准 JSON 格式的响应数据）
|
||
|
||
### 2.1 查询辅助
|
||
|
||
由于 fastapi 框架下面并没有类似 Django REST Framework 这种开箱即用的支持分页、排序、过滤的封装，因此想要快速实现支持分页、排序、过滤的数据查询代码需要不少额外的工作，因此本节设计了 `BaseQueryParams` 辅助查询类，可以用于快速实现列表查询。使用示例如下：
|
||
|
||
#### 2.1.1 基础使用
|
||
|
||
```python
|
||
#######################################################################################
|
||
# model.py
|
||
#######################################################################################
|
||
from sqlalchemy import Column, Integer, String, DateTime
|
||
from sqlalchemy.sql import func
|
||
from app.database import Base
|
||
|
||
class Person(Base):
|
||
__tablename__ = "sys_person"
|
||
id = Column(Integer, primary_key=True)
|
||
name = Column(String(254), unique=True)
|
||
age = Column(Integer)
|
||
created_at = Column(DateTime(timezone=True), server_default=func.now())
|
||
|
||
#######################################################################################
|
||
# schemas.py
|
||
#######################################################################################
|
||
from pydantic import BaseModel
|
||
from datetime import datetime
|
||
|
||
class Person(BaseModel):
|
||
id: int
|
||
name: str
|
||
age: int
|
||
created_at: datetime
|
||
|
||
class Config:
|
||
orm_mode = True
|
||
|
||
#######################################################################################
|
||
# query.py
|
||
#######################################################################################
|
||
from typing import Optional
|
||
from app import models
|
||
from sysom_utils import BaseQueryParams
|
||
|
||
|
||
class QueryParams(BaseQueryParams):
|
||
|
||
# 1. 指定要查询的模型（数据库模型）
|
||
__modelclass__ = models.Person
|
||
|
||
# 2. 定义排序字段（这边定义按创建时间降序，如果期望按时间升序，直接使用 "created_at" 即可）
|
||
sort: str = "-created_at"
|
||
|
||
# 3. 定义支持用于过滤的参数
|
||
# - 支持在query参数里面使用 name=A 过滤名字为A的用户;
|
||
# - 需要过滤出多个用户可以用逗号连接，比如 name=A,B
|
||
# - Optional 表示这个过滤字段是可选的，不传不过滤
|
||
name: Optional[str] = None
|
||
# - 支持在query参数里面使用 age=18 过滤年龄为18的用户;
|
||
# - 需要过滤出多个年龄的用户可以用逗号连接，比如 name=18,28
|
||
# - Optional 表示这个过滤字段是可选的，不传不过滤
|
||
age: Optional[str] = None
|
||
|
||
|
||
#######################################################################################
|
||
# crud.py
|
||
#######################################################################################
|
||
from typing import Optional, List
|
||
from sqlalchemy.orm import Session
|
||
from app import models, schemas, query
|
||
|
||
def get_person_list(db: Session, query_params: query.PersonQueryParams) -> List[models.Person]:
|
||
return (
|
||
query_params.get_query_exp(db)
|
||
.all()
|
||
)
|
||
|
||
#######################################################################################
|
||
# routes/person.py
|
||
#######################################################################################
|
||
|
||
from fastapi import APIRouter, Depends
|
||
from sqlalchemy.orm import Session
|
||
from app.query import PersonQueryParams
|
||
from app.database import get_db
|
||
from app.crud import get_person_list
|
||
from app.schemas import Person
|
||
from sysom_utils import StandardListResponse
|
||
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
@router.get("/list")
|
||
async def get_persons(
|
||
query_params: PersonQueryParams = Depends(), db: Session = Depends(get_db)
|
||
):
|
||
person_list = get_person_list(db, query_params)
|
||
return StandardListResponse(person_list, schema_class=Person)
|
||
```
|
||
|
||
#### 2.1.1 查询列表的同时获取条目总数
|
||
|
||
查询数据列表时，通常会设置过滤条件、排序、分页，获取总的数据数量应该是过滤之后的数量，此时可以通过如下方式同时获取到数据列表和总数：
|
||
|
||
```python
|
||
#######################################################################################
|
||
# crud.py
|
||
#######################################################################################
|
||
from typing import Optional, List, Tuple
|
||
from sqlalchemy.orm import Session
|
||
from app import models, schemas, query
|
||
|
||
def get_person_list(db: Session, query_params: query.PersonQueryParams) -> Tuple[List[models.Person], int]:
|
||
query_builder = query_params.get_query_builder(db)
|
||
|
||
# 1. Get total count after apply filter
|
||
query_builder.apply_filter()
|
||
total_count = len(query_builder.build().all())
|
||
|
||
# 2. Get alert list after apply sorter and paging
|
||
alert_list = query_builder.apply_sorter().apply_paging().build().all()
|
||
|
||
return alert_list, total_count
|
||
```
|
||
|
||
### 2.2 响应辅助
|
||
|
||
`sysom_utils` 设计了 `StandardReponse` 和 `StandardListResponse` 分别代表对象类型的响应数据和列表类型的响应数据，建议使用这两个响应对象封装 JSON 格式的响应数据。
|
||
|
||
#### 2.2.1 StandardReponse 使用示例
|
||
|
||
```python
|
||
from fastapi import APIRouter, Depends
|
||
from sqlalchemy.orm import Session
|
||
from app.database import get_db
|
||
from app.crud import get_person_by_name
|
||
from app.schemas import Person
|
||
from sysom_utils import StandardResponse
|
||
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
@router.get("/get")
|
||
async def get_specific_person(
|
||
person_name: str, db: Session = Depends(get_db)
|
||
):
|
||
person = get_person_by_name(db, person_name)
|
||
return StandardResponse(person, schema_class=Person)
|
||
|
||
#######################################################################################
|
||
# 响应数据的格式如下
|
||
# {
|
||
# "code": 200, => 状态码，200 表示成功，500表示服务端错误
|
||
# "message": "", => 错误信息，如果请求不成功这边可以存放错误
|
||
# "data": { => 字典类型的响应数据
|
||
# "id": 1,
|
||
# "name": "test",
|
||
# "age": 18
|
||
# }
|
||
# }
|
||
#
|
||
# 成功响应实例 => StandardResponse.success(data=person, schema_class=Person) 或者 StandardResponse(code=200, data=person, schema_class=Person)
|
||
# 错误响应实例 => StandardResponse.error(message="Username not exists") 或者 StandardResponse(code=500, message="Username not exists")
|
||
#######################################################################################
|
||
```
|
||
|
||
#### 2.2.1 StandardListReponse 使用示例
|
||
|
||
```python
|
||
from fastapi import APIRouter, Depends
|
||
from sqlalchemy.orm import Session
|
||
from app.query import PersonQueryParams
|
||
from app.database import get_db
|
||
from app.crud import get_person_list
|
||
from app.schemas import Person
|
||
from sysom_utils import StandardListResponse
|
||
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
@router.get("/list")
|
||
async def get_persons(
|
||
query_params: PersonQueryParams = Depends(), db: Session = Depends(get_db)
|
||
):
|
||
person_list = get_person_list(db, query_params)
|
||
return StandardListResponse(person_list, schema_class=Person)
|
||
|
||
#######################################################################################
|
||
# 响应数据的格式如下
|
||
# {
|
||
# "code": 200, => 状态码，200 表示成功，500表示服务端错误
|
||
# "message": "", => 错误信息，如果请求不成功这边可以存放错误
|
||
# "data": [ => 列表类型的响应数据
|
||
# {
|
||
# "id": 1,
|
||
# "name": "test",
|
||
# "age": 18
|
||
# }
|
||
# ]
|
||
# }
|
||
#
|
||
# 成功响应实例 => StandardResponse.success(data=person_list, schema_class=Person) 或者 StandardResponse(code=200, data=person_list, schema_class=Person)
|
||
# 错误响应实例 => StandardResponse.error(message="Username not exists") 或者 StandardResponse(code=500, message="Username not exists")
|
||
#######################################################################################
|
||
```
|