110 lines
3.6 KiB
Python
110 lines
3.6 KiB
Python
# 导入 SQLAlchemy 的 Session 类，用于类型提示
|
|
from sqlalchemy.orm import Session
|
|
# 导入地区维度表模型
|
|
from model.student_info_region_model import Region_dimension
|
|
|
|
|
|
# ==================== 查询地区 ====================
|
|
def get_region_part(session: Session, province: str):
|
|
"""
|
|
根据省份查询地区信息。
|
|
:param session: 数据库会话
|
|
:param province: 省份名称，若为空则查询全部
|
|
:return: Region_dimension 对象列表
|
|
"""
|
|
try:
|
|
# 如果传入了省份（非空），则按省份过滤
|
|
if province:
|
|
data = session.query(Region_dimension).filter(
|
|
Region_dimension.province == province
|
|
).all()
|
|
else:
|
|
# 否则查询所有地区
|
|
data = session.query(Region_dimension).all()
|
|
return data
|
|
except Exception as e:
|
|
# 出现异常直接向上抛出，由 API 层统一处理
|
|
raise e
|
|
|
|
|
|
# ==================== 添加地区 ====================
|
|
def add_region(session: Session, d1: dict):
|
|
"""
|
|
新增一条地区记录。
|
|
:param session: 数据库会话
|
|
:param d1: 地区信息字典，需包含 code、province、city、district
|
|
:return: 成功时返回字符串 '添加成功'
|
|
"""
|
|
try:
|
|
# 先检查该地区编码是否已存在
|
|
data = session.query(Region_dimension).filter(
|
|
Region_dimension.code == d1["code"]
|
|
).first()
|
|
|
|
# 如果已存在，抛出 ValueError（注意：这里没有带错误信息，调用方需自行处理）
|
|
if data:
|
|
raise ValueError
|
|
|
|
# 根据字典创建 ORM 对象
|
|
obj = Region_dimension(**d1)
|
|
session.add(obj) # 加入会话
|
|
session.commit() # 提交事务，写入数据库
|
|
session.refresh(obj) # 刷新对象，获取数据库生成的默认值（如自增 id）
|
|
return '添加成功'
|
|
except Exception as e:
|
|
# 出现异常时回滚事务，避免脏数据
|
|
session.rollback()
|
|
raise e
|
|
|
|
|
|
# ==================== 修改地区 ====================
|
|
def update_region(session: Session, code, d1: dict):
|
|
"""
|
|
根据地区编码更新地区信息（部分字段更新）。
|
|
:param session: 数据库会话
|
|
:param code: 要修改的地区编码
|
|
:param d1: 待更新的字段字典
|
|
:return: 成功返回 '修改成功'；未找到记录返回 None
|
|
"""
|
|
try:
|
|
# 使用 query.update() 批量更新，返回受影响行数
|
|
rows = session.query(Region_dimension).filter(
|
|
Region_dimension.code == code
|
|
).update(d1)
|
|
|
|
# 如果受影响行数为 0，说明编码不存在，无记录可更新
|
|
if rows == 0:
|
|
return None
|
|
|
|
session.commit()
|
|
return '修改成功'
|
|
except Exception as e:
|
|
# 异常时回滚
|
|
session.rollback()
|
|
raise e
|
|
|
|
|
|
# ==================== 删除地区 ====================
|
|
def delete_region(session: Session, code):
|
|
"""
|
|
根据地区编码删除地区记录（物理删除）。
|
|
:param session: 数据库会话
|
|
:param code: 要删除的地区编码
|
|
:return: 成功返回 '删除成功'；未找到记录返回 None
|
|
"""
|
|
try:
|
|
# 使用 query.delete() 物理删除，返回受影响行数
|
|
rows = session.query(Region_dimension).filter(
|
|
Region_dimension.code == code
|
|
).delete()
|
|
|
|
# 如果受影响行数为 0，说明编码不存在
|
|
if rows == 0:
|
|
return None
|
|
|
|
session.commit()
|
|
return '删除成功'
|
|
except Exception as e:
|
|
# 异常时回滚
|
|
session.rollback()
|
|
raise e |