SQLAlchemy的使用

from coreschema import Integer
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index



Base = declarative_base()

class Users(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False, index=True)
    depart_id = Column(Integer)



# 可以将 engine 提取出来共用
engine = create_engine(
        "mysql+pymysql://root:123@127.0.0.1:3306/peiqi_120?charset=utf8",
        max_overflow=0,     # 超过连接池大小外最多创建的连接
        pool_size=5,        # 连接池大小
        pool_timeout=10,    # 池中没有线程时最多等待的时间，否则报错
        pool_recycle=-1     # 多久之后对线程池中的线程进行一次连接的回收（重置）
    )


def create_tb():
    """
    创建表
    :return: 
    """
    engine = create_engine(
        "mysql+pymysql://root:123@127.0.0.1:3306/peiqi_120?charset=utf8",
        max_overflow=0,     # 超过连接池大小外最多创建的连接
        pool_size=5,        # 连接池大小
        pool_timeout=10,    # 池中没有线程时最多等待的时间，否则报错
        pool_recycle=-1     # 多久之后对线程池中的线程进行一次连接的回收（重置）,-1表示不重置
    )
    Base.metadata.create_all(engine)    # create_all() 必须是这个函数名, 表示创建表


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

# 执行操作
create_tb()
drop_tb()

from sqlalchemy import create_engine, and_, or_
from sqlalchemy.orm import sessionmaker
from models import Users



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

SessionFactory = sessionmaker(bind=engine)


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

###################### 单表的 crud 操作
‘‘‘
# 1 增加
user = Users(name="张三")
session.add(user)
session.commit()

# 批量增加
session.add_all([
    Users(name="李四"),
    Users(name="王五"),
    Users(name="赵六"),
])
session.commit()


# 2 查找
result = session.query(Users).all()
print(result)   #  [Users object, Users object, Users object, .....]
for item in result:
    print(item.id, "----",item.name)

result = session.query(Users).filter(Users.id >= 3)
print(result)   # sql 语句
for i in result:
    print(i.name)

result = session.query(Users).filter(Users.id >= 3).first()
print(result.name)


# 3 删除
ret = session.query(Users).filter(Users.id > 3).delete()
print(ret)  # 1 返回删除的数据量
session.commit()


# 4 修改
ret = session.query(Users).filter(Users.id == 1).update({Users.name:"张三三"})
print(ret)  # 1 返回修改的数据量
session.commit()

session.query(Users).filter(Users.id == 2).update({"name":"李思思"})
session.commit()

session.query(Users).filter(Users.id == 3).update({"name":Users.name+"老五"}, synchronize_session=False)
                                                    # 当要拼接字符串时候, 需要加参数 synchronize_session=False
                                                    # 数字相加时, 不需要这个参数
session.commit()


###################### 单表的其他操作
# 1 获取指定字段信息
ret = session.query(Users.id,Users.name,Users.name.label("nick_name")).all()
print(ret)
for i in ret:
    print(i[0],i.name,i.nick_name)

#2 and, 两种方式一样
ret = session.query(Users).filter(Users.id==2,Users.name=="李思思").all()
print(ret)
ret = session.query(Users).filter(and_(Users.id==2,Users.name=="李思思")).all()
print(ret)

# 3 or
ret = session.query(Users).filter(or_(Users.id==1,Users.id==2)).all()
print(ret)

# and 和 or 结合使用
ret = session.query(Users).filter(or_(
    Users.id == 1,
    and_(Users.name=="李思思", Users.id>=1),
    )).all()
print(ret)

# 4 between, 闭区间, 包括前后范围(1,3)
ret = session.query(Users).filter(Users.id.between(1,3)).all()
print(ret)

# 5 in 和 ~in(就是not in)
ret = session.query(Users).filter(Users.id.in_([1,2,3,4])).all()
print(ret)

ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
print(ret)

# 6 子查询
ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter(or_(Users.id>2, Users.name=="张三三")))).all()
print(ret)

# 7 filter_by,  里面是参数, 不是语句(比如 name == "李思思")
ret = session.query(Users).filter_by(name="李思思").all()
print(ret)

# 8 通配符 ( % :表示0个或多个,  _ :表示一个)
ret = session.query(Users).filter(Users.name.like("李%")).all()
print(ret)
ret = session.query(Users).filter(~Users.name.like("李%")).all()
print(ret)
ret = session.query(Users).filter(Users.name.like("李__")).all()
print(ret)

# 9 切片
ret = session.query(Users).filter(Users.id>=1)[1:3]
print(ret)

# 10 order_by 排序
ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()

# 11 group_by 分组
from sqlalchemy.sql import func
ret = session.query(
    Users.depart_id,
    func.count(Users.id)
    ).group_by(Users.depart_id).all()
for i in ret:
    print(i)

# group_by 和 having 结合使用
from sqlalchemy.sql import func
ret = session.query(
    Users.id,
    func.count(Users.id)
    ).group_by(Users.depart_id).having(func.count(Users.id)>=2).all()
print(ret)
for i in ret:
    print(i)

# 12 union 和 union_all

# select id,name from users UNION select id,name from users;

q1 = session.query(Users.name).filter(Users.id > 3)
q2 = session.query(Users.name).filter(Users.id >= 1)
ret = q1.union(q2).all()    # union() 去重
print(ret)  # [(‘赵六‘,), (‘张三三‘,), (‘李思思‘,), (‘王五老五‘,)]

q1 = session.query(Users.name).filter(Users.id > 3)
q2 = session.query(Users.name).filter(Users.id >= 1)
ret = q1.union_all(q2).all()    # union_all() 不去重
print(ret)  # [(‘赵六‘,), (‘张三三‘,), (‘李思思‘,), (‘王五老五‘,), (‘赵六‘,)]
‘‘‘