sysom1/template

微服务模板

1. 快速新建微服务

# 在 sysom/script 下调用如下命令
./sysom.sh create server <微服务名称> <微服务监听端口>

注意：

• 微服务名称不用带 sysom_ 前缀，会自动补充；
• 微服务的名称不能和现有的微服务冲突 => 如果冲突创建过程会报错提示
• 微服务监听端口不能和现有的微服务冲突 => 如果冲突创建过程会报错提示

比如创建一个名称为 demo 的微服务可以执行下列命令：

./sysom.sh create server demo 7010

执行后的目录结果如下：

├── conf
│   └── config.yml
├── cookie
├── deps
├── docker
├── docs
├── environment
├── LICENSE
├── package_rpm_offline.sh
├── package_rpm_online.sh
├── package.sh
├── README.md
├── republish.sh
├── script
│   ├── deploy
    ├── server
    │   ├── clear_exclude
    │   ├── clear.sh
    │   ├── deploy_exclude
    │   ├── init.sh
    │   ├── sysom_api
    │   ├── sysom_channel
    │   ├── sysom_demo                 => 新增 demo 微服务的部署脚本
    │   ├── sysom_diagnosis
    │   ├── sysom_hotfix
    │   ├── sysom_hotfix_builder
    │   ├── sysom_migration
    │   ├── sysom_monitor_server
    │   ├── sysom_vmcore
    │   └── sysom_vul
│   ├── sysom_clear.sh
│   ├── sysom_create.sh
│   ├── sysom_dbmigrate.sh
│   ├── sysom_deploy.sh
│   └── sysom.sh
├── sysom_server
│   ├── clear_exclude
│   ├── conf
│   ├── deploy_exclude
│   ├── sysom_api
│   ├── sysom_channel
│   ├── sysom_demo                      => 新增 demo 微服务目录
│   ├── sysom_diagnosis
│   ├── sysom_hotfix
│   ├── sysom_hotfix_builder
│   ├── sysom_migration
│   ├── sysom_monitor_server
│   ├── sysom_vmcore
│   └── sysom_vul
├── sysom_web
├── template

2. fastapi_helper

fastapi_helper 对使用 fastapi 编写微服务的一些常用操作进行了封装，可以提供如下开箱即用的能力：

• 查询辅助（支持分页、排序、按字段过滤等等）
• 响应辅助（支持快速返回标准 JSON 格式的响应数据）

2.1 查询辅助

由于 fastapi 框架下面并没有类似 Django REST Framework 这种开箱即用的支持分页、排序、过滤的封装，因此想要快速实现支持分页、排序、过滤的数据查询代码需要不少额外的工作，因此本节设计了 BaseQueryParams 辅助查询类，可以用于快速实现列表查询。使用示例如下：

2.1.1 基础使用

#######################################################################################
# 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 查询列表的同时获取条目总数

查询数据列表时，通常会设置过滤条件、排序、分页，获取总的数据数量应该是过滤之后的数量，此时可以通过如下方式同时获取到数据列表和总数：

#######################################################################################
# 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 使用示例

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 使用示例

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")
#######################################################################################