單表操作

阿新 • • 發佈：2021-11-06

單表操作

一、建立表

# 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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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:[email protected]: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)