单表操作

单表操作

一、创建表

# model.py
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()


# make_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)

    def __repr__(self):
        return self.name


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

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

    Base.metadata.drop_all(engine)


if __name__ == '__main__':
    # 创建表
    init_db()

    # 删除表
    # drop_db()

二、各种条件查询

2.1filter条件查询

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from model import Users

engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/db_flask", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()

# 1 filter条件查询
# 多个条件是 and的关系
filter_by = session.query(Users).filter_by(name='randysun', age=18).all()
filter_by1 = session.query(Users).filter_by(name='randysun').all()

print(filter_by)
print(filter_by1)

# 表达式，and条件连接
ret1 = session.query(Users).filter(Users.id > 3, Users.name == 'randy').all()
print(ret1)

2.2 between查询

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from model import Users

engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/db_flask", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()

# 2. between查询

between = session.query(Users).filter(Users.id.between(1,3)).all()
between_sql = session.query(Users).filter(Users.id.between(1,3))
print(between)
print(between_sql)

2.3 in 查询

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from model import Users

engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/db_flask", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()

# 3. in查询，注意下划线
res_in = session.query(Users).filter(Users.id.in_([1, 2, 3])).all()
res_in_sql = session.query(Users).filter(Users.id.in_([1, 2, 3]))
print(res_in)
print(res_in_sql)

2.4 ~ 非查询

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from model import Users

engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/db_flask", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()

# 4. ~ 非,除了什么什么之外 相当于not

# 相当于 sql中 not in(1, 2, 3)
res_in = session.query(Users).filter(~ Users.id.in_([1, 2, 3])).all()
res_in_sql = session.query(Users).filter(~ Users.id.in_([1, 2, 3]))
print(res_in)
print(res_in_sql)

2.5 二次筛选查询

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from model import Users

engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/db_flask", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()

# 5. 二次筛选 相当于sql中 select name from user where id in (select id from user where name='randy)
res = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='randy'))).all()
res_sql = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='randy')))
print(res)
print(res_sql)

2.6 通配符查询

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from model import Users

engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/db_flask", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()

# 6 通配符查询 like

res = session.query(Users).filter(Users.name.like('_a%')).all()
res_sql = session.query(Users).filter(Users.name.like('_a%'))
print(res)
print(res_sql)

res = session.query(Users).filter(~Users.name.like('l%')).all()
res_sql = session.query(Users).filter(~Users.name.like('l%')).all()
print(ret)

2.7 区间查询

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from model import Users

engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/db_flask", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()

# 7 限制，区间查询,用于分页
res = session.query(Users)[0:10]
print(res)

2.8 排序

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from model import Users

engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/db_flask", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()

# 8  排序,
# 1. 根据id升序排序
res = session.query(Users.id, Users.name).order_by(Users.id.asc()).all()
print(res)

# 2. 根据id,降序排序
res = session.query(Users.id, Users.name).order_by(Users.id.desc()).all()
print(res)

# 3. 第一个条件重复后，再按第二个条件升序排
ret = session.query(Users.id, Users.name).order_by(Users.id.asc(), Users.name.desc()).all()
print(ret)

2.9 分组

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from model import Users

engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/db_flask", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()

# 10 分组

res = session.query(Users).group_by(Users.name).all()
print(res)

# 分组之后获取最大的id, 最小id, id之和
# 使用函数需要导入func模块
from sqlalchemy.sql import func

res = session.query(
    func.max(Users.id),
    func.min(Users.id),
    func.sum(Users.id),
    Users.name,
).group_by(Users.name).all()

print(res)

# haviing筛选
res = session.query(
    func.max(Users.id),
    func.min(Users.id),
    func.sum(Users.id),
    Users.name,
).group_by(Users.name).having(func.min(Users.id > 2)).all()
res_sql = session.query(
    func.max(Users.id),
    func.min(Users.id),
    func.sum(Users.id),
    Users.name,
).group_by(Users.name).having(func.min(Users.id > 2))

print(res)
print(res_sql)

三、and 和 or查询

导入模块from sqlalchemy import and_, or_

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from model import Users

engine = create_engine("mysql+pymysql://root:root@127.0.0.1:3306/db_flask", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()

from sqlalchemy import and_, or_

# or_包裹的都是or条件，and_包裹的都是and条件

# 1.and_
res = session.query(Users).filter(and_(Users.id > 3, Users.name == 'randy')).all()
res_sql = session.query(Users).filter(and_(Users.id > 3, Users.name == 'randy'))
print(res)
print(res_sql)

# 2. or_
res = session.query(Users).filter(or_(Users.id < 2, Users.name == 'randy')).all()
res_sql = session.query(Users).filter(or_(Users.id < 2, Users.name == 'randy'))
print(res)
print(res_sql)

# 3. or_ and_共同查询
res = session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == 'randy', Users.id > 3)
    )).all()
res_sql = session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == 'randy', Users.id > 3)
    ))
print(res)
print(res_sql)

四、总结

• filter系列查询：多个条件表示and关系
• between：区间查询，1,3两者之间
• in: 包含查询
• ~: 非查询， 相当于sql中 not
• and_和or_: 需要导入这连个模块，可以嵌套使用
• like: 模糊匹配， _表示任意单个字符， %表示任意多个字符
• order_by: 排序， asc升序，desc降序
• group_by: 分组查询，如果要使用聚合函数需要导入fun模块，分组之后查询条件使用 having