ExamService

题库小程序服务端 - FastAPI实现

项目结构

ExamService/
├── alembic/                    # 数据库迁移相关
├── mooc/                        # 主应用目录
│   ├── api/                    # API路由
│   │   └── v1/                # API v1版本
│   ├── core/                   # 核心配置
│   ├── crud/                   # 数据库操作
│   ├── db/                     # 数据库
│   ├── models/                 # 数据库模型
│   ├── schemas/               # Pydantic模型
│   └── utils/                 # 工具函数
├── tests/                     # 测试目录
└── [配置文件]

安装

1. 安装 Miniconda

在 Ubuntu 上安装 Miniconda，执行以下步骤：

# 下载 Miniconda 安装脚本
wget https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh

# 给安装脚本添加执行权限
chmod +x Miniconda3-latest-Linux-x86_64.sh

# 运行安装脚本
./Miniconda3-latest-Linux-x86_64.sh

# 按照提示完成安装，选择安装目录并接受许可协议

# 激活 Miniconda
source ~/.bashrc
# 创建名为 'mooc' 的虚拟环境，Python 版本大于等于 3.13
conda create --name mooc python>=3.13
# 激活 'mooc' 环境
conda activate mooc
# 安装依赖
pip install -r requirements.txt

安装mysql开发库

# 更新包列表
apt-get update

# 安装 MySQL 开发包和其他必要的包
apt-get install -y python3-dev default-libmysqlclient-dev build-essential pkg-config


pip install mysqlclient pymysql cryptography

运行

uvicorn main:app --reload

初始化数据库

alembic init alembic
alembic revision --autogenerate -m "Initial migration"
alembic upgrade head

添加数据库表

1.创建模型文件:

因为表格太多，可以把同类型表放在同一个文件，文件命名规则：ims_account_wechats、ims_account_wxapp、ims_account_xzapp等表格去除前缀和后缀，即为account.py

# 这个模型类AccountWechats用于映射ims_account_wechats表：
# 1. 在ORM层中代表数据库表结构，方便CRUD操作。
# 2. 在FastAPI中引用该模型时，可以使用Pydantic自动转换ORM对象到响应模型。
# 3. 创建模型时需确保表名(__tablename__)与真实的数据库表名保持一致，否则查询和插入可能失败。
# 4. 字段类型要与数据库中定义的列类型严格匹配，避免出现不兼容或异常。
# 5. 通过Primary Key（acid）唯一标识记录，确保ORM能正确追踪和更新该对象。
# 6. config.orm_mode = True：允许直接把SQLAlchemy模型实例转换为Pydantic模型对象。


# filepath: ExamService/mooc/models/account.py
from sqlalchemy import Column, Integer, String, SmallInteger
from mooc.db.database import Base
# 类名命名规则： 去除ims前缀后采用 帕斯卡命名法，无连接符每个单词手写大写
# AccountWechats模型用于映射数据库表 ims_account_wechats
class AccountWechats(Base):
    __tablename__ = "ims_account_wechats"

    acid = Column(Integer, primary_key=True)
    uniacid = Column(Integer, nullable=False)
    token = Column(String(32), nullable=False)
    encodingaeskey = Column(String(255), nullable=False)
    level = Column(SmallInteger, nullable=False)
    name = Column(String(30), nullable=False)
    account = Column(String(30), nullable=False)
    original = Column(String(50), nullable=False)
    signature = Column(String(100), nullable=False)
    country = Column(String(10), nullable=False)
    province = Column(String(3), nullable=False)
    city = Column(String(15), nullable=False)
    username = Column(String(30), nullable=False)
    password = Column(String(32), nullable=False)
    lastupdate = Column(Integer, nullable=False)
    key = Column(String(50), nullable=False)
    secret = Column(String(50), nullable=False)
    styleid = Column(Integer, nullable=False)
    subscribeurl = Column(String(120), nullable=False)
    auth_refresh_token = Column(String(255), nullable=False)

    class Config:
        orm_mode = True # 允许与Pydantic的ORM功能兼容，直接读取数据库模型对象

2.创建schema:

# filepath: ExamService/mooc/schemas/account.py

from pydantic import BaseModel
from typing import Optional

class AccountWechatsBase(BaseModel):
    """
    数据模型基类: AccountWechatsBase
    用于描述基础字段的类型、用途和注意点。
    - 注意: 必填字段都在这里声明，每个字段都对应数据库中的一列。
    """
    uniacid: int            # 微信公众号/小程序的关联ID
    token: str              # 访问接口时使用的token
    encodingaeskey: str     # 用于消息加解密的AES密钥
    level: int              # 认证等级，如订阅号/服务号
    name: str               # 微信公众号名称
    account: str            # 公众号帐号，如微信号
    original: str           # 原始ID(通常以gh_开头)
    signature: str          # 公众号信息签名
    country: str            # 国家
    province: str           # 省份
    city: str               # 城市
    username: str           # 后台登录用户名
    password: str           # 后台登录密码
    lastupdate: int         # 最后一次更新的时间戳
    key: str                # 开发者Key
    secret: str             # 开发者Secret
    styleid: int            # 模板样式ID
    subscribeurl: str       # 订阅链接
    auth_refresh_token: str # 用来刷新授权的token

class AccountWechatsCreate(AccountWechatsBase):
    """
    用于创建新微信帐号记录:
    - 继承自AccountWechatsBase, 不额外添加字段
    - 仅表示此Schema专用于'创建'场景
    """

    # 用于创建新记录，包含所有必填字段
    pass

class AccountWechatsUpdate(BaseModel):
    """
    用于更新已有微信帐号记录:
    - 只包含可选字段，未在此处的内容将保持不变
    - 注意: exclude_unset=True 可以避免更新空值
    """
    token: Optional[str]         # 可选更新token
    encodingaeskey: Optional[str]# 可选更新AES key

class AccountWechats(AccountWechatsBase):
    """
    表示完整的微信帐号记录:
    - acid: 数据库主键ID
    - 包含所有字段的最终模型，ORM转换时使用
    """
    acid: int              # 表中的主键ID

    class Config:
        orm_mode = True    # 允许与ORM对象进行直接转换

3.创建 CRUD:

# filepath: ExamService/mooc/crud/crud_account.py
from sqlalchemy.orm import Session
from typing import Optional

from mooc.models.account import AccountWechats
from mooc.schemas.account import AccountWechatsCreate, AccountWechatsUpdate

class CRUDAccountWechats:
    """
    创建CRUD
    负责对AccountWechats模型进行增删改查操作的逻辑封装。
    """

    def create(self, db: Session, obj_in: AccountWechatsCreate) -> AccountWechats:
        """
        创建记录：将Pydantic的输入数据转换为数据库模型实例并保存。
        """
        db_obj = AccountWechats(**obj_in.dict())  # 将表单数据解包到模型实例
        db.add(db_obj)                           # 添加到会话
        db.commit()                              # 提交事务
        db.refresh(db_obj)                       # 刷新实例，获取数据库中的最新状态
        return db_obj

    def get(self, db: Session, acid: int) -> Optional[AccountWechats]:
        """
        根据主键 acid 查询单条数据。
        """
        return db.query(AccountWechats).filter(AccountWechats.acid == acid).first()

    def update(
        self, db: Session, *, db_obj: AccountWechats, obj_in: AccountWechatsUpdate
    ) -> AccountWechats:
        """
        更新记录：只修改传递进来的字段，未设置的字段不动。
        """
        for field, value in obj_in.dict(exclude_unset=True).items():
            setattr(db_obj, field, value)  # 更新模型实例属性
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj

    def delete(self, db: Session, acid: int) -> None:
        """
        删除记录：物理删除或逻辑删除都可在此处实现。
        """
        obj = db.query(AccountWechats).filter(AccountWechats.acid == acid).first()
        if obj:
            db.delete(obj)
            db.commit()

# 实例化CRUD对象，方便在业务代码中直接引用
account_wechats = CRUDAccountWechats()

4.注册模型:

# filepath: ExamService/mooc/db/base.py
from mooc.db.database import Base
from mooc.models.admin import Admin, Account, AccountWebapp
from mooc.models.account import AccountWechats  # 新增,将创建好的模型类导入即可

5.运行代码更新数据库信息

python main.py

API文档

启动服务后访问: http://localhost:2333/docs