封装sql提交和回滚

class SQLAlchemy(_SQLAlchemy):
    @contextmanager
    def auto_commit(self):
        try:
            yield
            self.session.commit()
        except Exception as e:
            self.session.rollback()
            raise ServerError()

更新

with db.auto_commit():
    word.count += 1

添加

with db.auto_commit():
    word = HotWords()
    word.sid = stock.id
    db.session.add(word)

创建表与表关联（外键）

class Optional(Base):
    __tablename__ = 'optional'

    id = Column(Integer, primary_key=True)
    uid = Column(Integer, ForeignKey('user.id'), nullable=False)
    sid = Column(Integer, ForeignKey('stock.id'), nullable=False)
    user = relationship('User')
    stock = relationship('Stock')

实现三表查询

result = db.session.query(Optional.sid, Stock.name, Stock.ts_code, Stock.symbol)\
    .join(User, Optional.uid == User.id)\
    .join(Stock, Optional.sid == Stock.id).filter(User.openid == openid).all()

分组查询 / 排序查询

db.session.query(Stock.id, Stock.ts_code, Stock.symbol, Stock.name, Stock.area)\
    .group_by(Stock.area).all()
db.session.query(Stock.id, Stock.ts_code, Stock.symbol, Stock.name, Stock.area)\
    .order_by(Stock.area).all()

实现批量新增数据

插入数据不多时倒是可以通过循环解决，但是一旦数量多就很慢了

# db是sqlalchemy对象，ExamSchool是通过db.Modal创建的数据表，虽然也是循环新增，不过效率快多了
db.session.execute(
	ExamSchool.__table__.insert(),
	[{"eid": int(form.get('id')), "sid": int(i)} for i in new_schools]
)

实现批量更新数据

# 一种，ExamSchool与exam是一对多关系
db.session.query(ExamSchool).filter(ExamSchool.eid == exam.id). \
                update({ExamSchool.status: 0}, synchronize_session=False)
# 第二种
from sqlalchemy import bindparam
update_data = [{'b_id': i, 'status': 0} for i in old_schools.values()]
stmt = ExamSchool.__table__.update().where(ExamSchool.id == bindparam('b_id')).values({ExamSchool.status: bindparam('status')})
db.session.execute(stmt, update_data)

实现批量删除数据

# 注意这是物理删除，不是逻辑删除
db.session.query(ExamSchool).filter(ExamSchool.eid == exam.id).delete()