SQLAlchemy ORM

# pip install  sqlalchemy
import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
from sqlalchemy.orm import relationship
Base = declarative_base()

class Users(Base):
    __tablename__ = 'users'  # 数据库表名称
    id = Column(Integer, primary_key=True)  # id 主键
    name = Column(String(32), index=True, nullable=False)  # name列，索引，不可为空
    age = Column(Integer)
    #email = Column(String(32), unique=True)
    #datetime.datetime.now#不能加括号，加了括号，以后永远是当前时间
    #ctime = Column(DateTime, default=datetime.datetime.now)
    #extra = Column(Text, nullable=True)

    __table_args__ = (
        # UniqueConstraint('id', 'name', name='uix_id_name'), #联合唯一
        # Index('ix_id_name', 'name', 'email'), #索引
    )

    def __repr__(self):  # 作用类似于__str__, 但是__repr__的作用更加底层，__str__则打印不出来
        return self.name


def init_db():
    """
    根据类创建数据库表
    :return:
    """
    engine = create_engine(
        "mysql+pymysql://root:123456@127.0.0.1:3306/python13?charset=utf8",
        #"什么数据库（mysql,orcal）+用什么取链接数据库（pymysql）://数据库用户名:密码@mysqlip:端口/数据库名？charset=字符集"
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
    )

    Base.metadata.create_all(engine)

def drop_db():
    """
    根据类删除数据库表
    :return:
    """
    engine = create_engine(
         "mysql+pymysql://root:123456@127.0.0.1:3306/python13?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
    )

    Base.metadata.drop_all(engine)

if __name__ == '__main__':
    drop_db()
    init_db()

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Users
#"mysql+pymysql://root@127.0.0.1:3306/aaa"
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/python13", max_overflow=0, pool_size=5)
Connection = sessionmaker(bind=engine)

# 每次执行数据库操作时，都需要创建一个Connection
session = Connection()

# 条件
ret =  session.query(Users).filter_by(name = "esb").all()

#表达式，and 条件链接
ret  = session.query(Users).filter(Users.name == "sb",Users.age ==14).first()
# print(ret.age,ret.name)

# 表示的between（包括头尾），条件,30<=age<=40
ret = session.query(Users).filter(Users.age.between(30,40)).all()
# print(ret)


# sql查询的in_操作,相当于django中的__in
ret =session.query(Users).filter(Users.id.in_([9,11,13])).all()
# print(ret)


# # sql查询取反
ret1 = session.query(Users).filter(~Users.id.in_([9,11,13])).all()
# print(ret1)


#or查询 ，or和and ,做整合
# 想要使用，的向导入
from sqlalchemy import or_, and_
#
ret = session.query(Users).filter(or_(Users.id == 9, Users.name=="jsb")).all()
ret = session.query(Users).filter(and_(Users.id == 9,Users.name=="lsb1")).all()

# 也可以包裹在一起使用
ret =  session.query(Users).filter(or_(
    Users.id == 9,
    and_(Users.name=="jsb",Users.id==13),

    )
).all()


# like查询,模糊配置
#必须以b开头
ret = session.query(Users).filter(Users.name.like("b%")).all()

# #第二字母是b
ret = session.query(Users).filter(Users.name.like("_b%")).all()

#不以b开头
ret = session.query(Users).filter(~Users.name.like("b%")).all()


#排序
#desc从大到小排序
ret = session.query(Users).filter(Users.id>1).order_by(Users.id.desc()).all()

#asc从小到大排序
ret = session.query(Users).filter(Users.id>1).order_by(Users.id.asc()).all()

#多条件排序,先以年纪从大到小排，如果年龄相同，再以id从小到大排
ret = session.query(Users).filter(Users.id>1).order_by(Users.age.desc(),Users.id.asc()).all()
# print(ret)


#分组查询
ret  = session.query(Users).group_by(Users.name).all()

# 再分组的时候如果要用聚合操作，就要导入func
from sqlalchemy.sql import func  # func 就类似于django的聚合查询

#选出组内最小年龄要大于等于30的组
ret  = session.query(Users).group_by(Users.name).having(func.min(Users.age)>=30).all()

#选出组内最小年龄要大于等于30的组,查询组内的最小年龄，最大年纪，年纪之和，
ret = session.query(
    func.min(Users.age),
    func.max(Users.age),
    func.sum(Users.age),
    Users.name
).group_by(Users.name).having(func.min(Users.age)>=30).all()
print(ret)

@app.route("/sqlalchemy/test", methods=['GET'])
def sqlalchemy_test_api():
    data = {}
    # 获取商品价格
    product = Product.query.get(1)
    data['old_price'] = product.present_price
    # 休眠10秒，等待外部修改价格
    time.sleep(10)
    product = Product.query.get(1)
    data['new_price'] = product.present_price
    return jsonify(status='ok', data=data)

>>> from sqlalchemy import inspect
>>> insp = inspect(product)
>>> insp.persistent
True
>>> product.__dict__
{
  'id': 1, 'present_price': 10,
  '_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x1106a3350>,
}

db.session.expire(product)
db.session.refresh(product)
db.session.expire_all()
db.session.flush()
db.session.commit()