from flask_sqlalchemy import SQLAlchemy from sqlalchemy.orm import selectinload db SQLAlchemy() class BaseModel(db.Model): __abstract__ True # 插入相关 classmethod def insert(cls, data, ignore_extraTrue): 插入单条数据 :param data: 字典如 {name: 张三, age: 18} :param ignore_extra: 是否忽略字典中多余的字段True忽略False报错 :return: 保存后的模型实例 if ignore_extra: # 安全模式只取模型中存在的字段 instance cls() for key, value in data.items(): if hasattr(instance, key): setattr(instance, key, value) else: # 严格模式直接解包多余字段会报错 instance cls(**data) instance.save() return instance classmethod def insert_all(cls, data_list, ignore_extraTrue): 批量插入多条数据 :param data_list: 字典列表如 [{name: 张三}, {name: 李四}] :param ignore_extra: 是否忽略字典中多余的字段 :return: 保存后的模型实例列表 instances [] for data in data_list: if ignore_extra: instance cls() for key, value in data.items(): if hasattr(instance, key): setattr(instance, key, value) else: instance cls(**data) instances.append(instance) db.session.add_all(instances) db.session.commit() return instances # 转字典相关 def to_dict(self, excludeNone, includeNone): 将单个模型实例转为字典 :param exclude: 要排除的字段列表如 [password, deleted_at] :param include: 要包含的字段列表优先级高于 exclude :return: 字典 exclude exclude or [] include include or [] result {} for column in self.__table__.columns: name column.name # 如果指定了 include只返回 include 里的字段 if include and name not in include: continue # 排除 exclude 里的字段 if name in exclude: continue # 获取值处理 datetime 等特殊类型 value getattr(self, name) if hasattr(value, isoformat): # datetime/date 转字符串 value value.isoformat() result[name] value return result classmethod def to_dicts(cls, instances, excludeNone, includeNone): 将模型实例列表转为字典列表 :param instances: 模型实例列表通常来自 .all() :param exclude: 要排除的字段列表 :param include: 要包含的字段列表 :return: 字典列表 return [instance.to_dict(exclude, include) for instance in instances] # 统一查询 转字典最常用 classmethod def find(cls, id, excludeNone, includeNone): 根据 ID 查询并直接返回字典方便 API 输出 :param id: 主键 ID :param exclude: 排除的字段 :param include: 包含的字段 :return: 字典或 None instance cls.query.get(id) if instance: return instance.to_dict(exclude, include) return None classmethod def select(cls, excludeNone, includeNone, **filters): 查询列表并直接返回字典列表 :param exclude: 排除的字段 :param include: 包含的字段 :param filters: 查询条件如 name张三, age18 :return: 字典列表 query cls.query for key, value in filters.items(): query query.filter_by(**{key: value}) instances query.all() return cls.to_dicts(instances, exclude, include) # 你原有的方法 def save(self): db.session.add(self) db.session.commit() return self def delete(self): db.session.delete(self) db.session.commit() return self classmethod def get(cls, id): return cls.query.get(id) def update(self, data, ignore_extraTrue): 更新数据 :param data: 字典 :param ignore_extra: 是否忽略多余字段 for key, value in data.items(): if ignore_extra and not hasattr(self, key): continue setattr(self, key, value) return self.save()# 假设定义 User 模型 class User(BaseModel): __tablename__ users id db.Column(db.Integer, primary_keyTrue) name db.Column(db.String(50)) age db.Column(db.Integer) password db.Column(db.String(100)) # 敏感字段 # 插入 # 单条插入自动忽略多余字段 data {name: 张三, age: 18, extra_field: 会被忽略} user User.insert(data) # 返回 User 实例 # 批量插入 data_list [ {name: 李四, age: 20}, {name: 王五, age: 22} ] users User.insert_all(data_list) # 返回 [User, User] # 转字典 user User.get(1) # 转成全部字段 print(user.to_dict()) # {id: 1, name: 张三, age: 18, password: xxx} # 排除敏感字段 print(user.to_dict(exclude[password])) # {id: 1, name: 张三, age: 18} # 只包含指定字段 print(user.to_dict(include[name, age])) # {name: 张三, age: 18} # 查询直接返回字典API 最常用 # 根据 ID 查询直接返回字典 user_dict User.find(1, exclude[password]) # 返回: {id: 1, name: 张三, age: 18} # 条件查询直接返回字典列表 users_dict User.select(exclude[password], age18) # 返回: [{id: 1, name: 张三, age: 18}, ...] # 如果不想过滤查全部 all_users User.select()