sqlalchemy 基本操作

表操作

models.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column
from sqlalchemy import Integer,String,Text,Date,DateTime
from sqlalchemy import create_engine
 
 
Base = declarative_base()
 
class Users(Base):
    __tablename__ = 'users'
 
    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)
 
 
def create_all():
    engine = create_engine(
        "mysql+pymysql://root:123456@127.0.0.1:3306/s9day120?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
    )
 
    Base.metadata.create_all(engine)
 
def drop_all():
    engine = create_engine(
        "mysql+pymysql://root:123456@127.0.0.1:3306/s9day120?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
    )
    Base.metadata.drop_all(engine)
 
if __name__ == '__main__':
    create_all()

views.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Users
 
# 创建引擎
engine = create_engine(
        "mysql+pymysql://root:123456@127.0.0.1:3306/s9day120?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
    )
 
# 根据引擎创建session工厂
SessionFactory = sessionmaker(bind=engine)
 
# 用session工厂创建一个session对象
session = SessionFactory()
 
..........
# 根据Users类对users表进行增删改查
..........
 
# 关闭session
session.close()

 行操作

增

1
2
3
4
5
6
7
8
9
obj = Users(name='alex')
session.add(obj)
session.commit()
 
session.add_all([
        Users(name='小东北'),
        Users(name='龙泰')
])
session.commit()

删

1
2
session.query(Users).filter(Users.id >= 2).delete()
session.commit()

改

1
2
3
4
session.query(Users).filter(Users.id == 4).update({Users.name:'东北'})
session.query(Users).filter(Users.id == 4).update({'name':'小东北'})
session.query(Users).filter(Users.id == 4).update({'name':Users.name+"DSB"},synchronize_session=False)
session.commit()

查

1
2
3
4
5
6
7
8
9
10
result = session.query(Users).all()
for row in result:
        print(row.id,row.name)
 
result = session.query(Users).filter(Users.id >= 2)
for row in result:
        print(row.id,row.name)
 
result = session.query(Users).filter(Users.id >= 2).first()
print(result)

其他常用操作

  1. 指定查询列

    1
    2
    3
    result = session.query(Users.id,Users.name.label('cname')).all()
    for item in result:
            print(item[0],item.id,item.cname)

  2. 多个查询条件（默认and）

    1
    session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()

  3. between

    1
    session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()

  4. in

    1
    2
    session.query(Users).filter(Users.id.in_([1,3,4])).all()
    session.query(Users).filter(~Users.id.in_([1,3,4])).all(） # 非

  5. 子查询

    1
    session.query(Users).filter(Users.id.in_(session.query(Users.id).filter(Users.name=='eric'))).all()

  6. and和or

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    from sqlalchemy import and_, or_
    session.query(Users).filter(Users.id > 3, Users.name == 'eric').all()
    session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
    session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
    session.query(Users).filter(
        or_(
            Users.id < 2,
            and_(Users.name == 'eric', Users.id > 3),
            Users.extra != ""
        )).all()

  7. filter_by

    1
    session.query(Users).filter_by(name='alex').all()

  8. 通配符

    1
    2
    ret = session.query(Users).filter(Users.name.like('e%')).all()
    ret = session.query(Users).filter(~Users.name.like('e%')).all()

  9. 切片

    1
    result = session.query(Users)[1:2]

  10. 排序

    1
    2
    ret = session.query(Users).order_by(Users.name.desc()).all()
    ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()

  11. group by

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    ret = session.query(
            Users.depart_id,
            func.count(Users.id),
    ).group_by(Users.depart_id).all()
    for item in ret:
            print(item)
     
    from sqlalchemy.sql import func
     
    ret = session.query(
            Users.depart_id,
            func.count(Users.id),
    ).group_by(Users.depart_id).having(func.count(Users.id) >= 2).all()
    for item in ret:
            print(item)

  12. union和union all

    1
    2
    3
    4
    5
    6
    7
    q1 = session.query(Users.name).filter(Users.id > 2)
    q2 = session.query(Favor.caption).filter(Favor.nid < 2)
    ret = q1.union(q2).all()
     
    q1 = session.query(Users.name).filter(Users.id > 2)
    q2 = session.query(Favor.caption).filter(Favor.nid < 2)
    ret = q1.union_all(q2).all()

models.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column
from sqlalchemy import Integer,String,Text,Date,DateTime
from sqlalchemy import create_engine
 
 
Base = declarative_base()
 
class Users(Base):
    __tablename__ = 'users'
 
    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)
 
 
def create_all():
    engine = create_engine(
        "mysql+pymysql://root:123456@127.0.0.1:3306/s9day120?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
    )
 
    Base.metadata.create_all(engine)
 
def drop_all():
    engine = create_engine(
        "mysql+pymysql://root:123456@127.0.0.1:3306/s9day120?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
    )
    Base.metadata.drop_all(engine)
 
if __name__ == '__main__':
    create_all()

views.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Users
 
# 创建引擎
engine = create_engine(
        "mysql+pymysql://root:123456@127.0.0.1:3306/s9day120?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
    )
 
# 根据引擎创建session工厂
SessionFactory = sessionmaker(bind=engine)
 
# 用session工厂创建一个session对象
session = SessionFactory()
 
..........
# 根据Users类对users表进行增删改查
..........
 
# 关闭session
session.close()

 行操作

增

1
2
3
4
5
6
7
8
9
obj = Users(name='alex')
session.add(obj)
session.commit()
 
session.add_all([
        Users(name='小东北'),
        Users(name='龙泰')
])
session.commit()

删

1
2
session.query(Users).filter(Users.id >= 2).delete()
session.commit()

改

1
2
3
4
session.query(Users).filter(Users.id == 4).update({Users.name:'东北'})
session.query(Users).filter(Users.id == 4).update({'name':'小东北'})
session.query(Users).filter(Users.id == 4).update({'name':Users.name+"DSB"},synchronize_session=False)
session.commit()

查

1
2
3
4
5
6
7
8
9
10
result = session.query(Users).all()
for row in result:
        print(row.id,row.name)
 
result = session.query(Users).filter(Users.id >= 2)
for row in result:
        print(row.id,row.name)
 
result = session.query(Users).filter(Users.id >= 2).first()
print(result)

其他常用操作

  1. 指定查询列

    1
    2
    3
    result = session.query(Users.id,Users.name.label('cname')).all()
    for item in result:
            print(item[0],item.id,item.cname)

  2. 多个查询条件（默认and）

    1
    session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()

  3. between

    1
    session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()

  4. in

    1
    2
    session.query(Users).filter(Users.id.in_([1,3,4])).all()
    session.query(Users).filter(~Users.id.in_([1,3,4])).all(） # 非

  5. 子查询

    1
    session.query(Users).filter(Users.id.in_(session.query(Users.id).filter(Users.name=='eric'))).all()

  6. and和or

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    from sqlalchemy import and_, or_
    session.query(Users).filter(Users.id > 3, Users.name == 'eric').all()
    session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
    session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
    session.query(Users).filter(
        or_(
            Users.id < 2,
            and_(Users.name == 'eric', Users.id > 3),
            Users.extra != ""
        )).all()

  7. filter_by

    1
    session.query(Users).filter_by(name='alex').all()

  8. 通配符

    1
    2
    ret = session.query(Users).filter(Users.name.like('e%')).all()
    ret = session.query(Users).filter(~Users.name.like('e%')).all()

  9. 切片

    1
    result = session.query(Users)[1:2]

  10. 排序

    1
    2
    ret = session.query(Users).order_by(Users.name.desc()).all()
    ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()

  11. group by

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    ret = session.query(
            Users.depart_id,
            func.count(Users.id),
    ).group_by(Users.depart_id).all()
    for item in ret:
            print(item)
     
    from sqlalchemy.sql import func
     
    ret = session.query(
            Users.depart_id,
            func.count(Users.id),
    ).group_by(Users.depart_id).having(func.count(Users.id) >= 2).all()
    for item in ret:
            print(item)

  12. union和union all

    1
    2
    3
    4
    5
    6
    7
    q1 = session.query(Users.name).filter(Users.id > 2)
    q2 = session.query(Favor.caption).filter(Favor.nid < 2)
    ret = q1.union(q2).all()
     
    q1 = session.query(Users.name).filter(Users.id > 2)
    q2 = session.query(Favor.caption).filter(Favor.nid < 2)
    ret = q1.union_all(q2).all()

相关文章：

猜你喜欢
相关资源
相似解决方案
热门标签
Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode