欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识！
您现在的位置是： 首页

【Python SQLAlchemy】

程序员文章站 2022-03-11 19:01:41
...

Python 的 ORM 框架 SQLAlchemy 有些同学已经听说过这个框架的大名了，也听说了 SQLAlchemy 没有 Django 的 models 好用。

Django 的 models 只是配置和使用比较简单，因为它是 Django 自带的 ROM 框架，也正因为它是 Django 原生的，所以兼容性远远不如 SQLAlchemy 。

真正算得上全面的 ORM 框架必然是我们的 SQLAlchemy 了，它可以在任何时候使用 SQL 查询时使用。

文章目录


单表操作

创建数据表 

from sqlalchemy.ext.declarative import declarative_base  # 导入官宣模型
from sqlalchemy import create_engine  # 用于创建数据库引擎
from sqlalchemy import Column  # 用于创建字段
from sqlalchemy import INT, Integer, INTEGER  # 这3个是一样的，使用其中一种即可
from sqlalchemy import String, VARCHAR  # varchar类型，使用其中一个即可


# 实例化官宣模型，Base就是ORM模型
Base = declarative_base()


# 创建数据表
class User(Base):  # 继承Base

    __tablename__ = 'user'  # 指定表名称

    # xx = Column(数据类型, 索引, 主键, 外键, 等等)

    id = Column(INT, primary_key=True, autoincrement=True)
    # primary_key=True：主键
    # autoincrement=True：自增，默认为True

    name = Column(String(32), index=True)
    # index=True：创建索引


# 创建数据库引擎
engine = create_engine('mysql+pymysql://root:[email protected]@localhost:3306/db01?charset=utf8')
# mysql+pymysql：使用的是pymysql
# root:[email protected]：用户名:密码
# localhost:3306：地址:端口
# db01：库名称
# charset=utf8：表示支持中文


# 将继承Base的所有的Class序列化成数据表
Base.metadata.create_all(engine)

增加数据 

# 用于创建sessionmaker会话对象
from sqlalchemy.orm import sessionmaker

# 导入我们创建的User类与数据库引擎engine
from blog.create_table import User, engine

# 创建sessionmaker会话对象，将数据库引擎engine交给sessionmaker
Session = sessionmaker(engine)

# 打开会话对象Session
db_session = Session()


"""添加单条数据"""

# 使用User ORM模型创建一条数据
user_obj = User(name='user01')

# 将创建的数据添加到会话对象中
db_session.add(user_obj)

# 提交数据
db_session.commit()
# commit方法是将db_session中的所有指令一次性提交


"""添加多条数据"""

# # 使用User ORM模型创建多条数据
user_list = [
    User(name='user02'),
    User(name='user03'),
    User(name='user04'),
]

# 将创建的所有数据添加到会话对象中
db_session.add_all(user_list)

# 提交
db_session.commit()


# 关闭会话对象
db_session.close()

查询数据 

# 用于创建sessionmaker会话对象
from sqlalchemy.orm import sessionmaker

# 导入我们创建的User类与数据库引擎engine
from blog.create_table import User, engine

from sqlalchemy.sql import and_, or_, text, func

# 创建sessionmaker会话对象，将数据库引擎engine交给sessionmaker
Session = sessionmaker(engine)

# 打开会话对象Session
db_session = Session()

# 指定查询的表（类）
user = db_session.query(User)


"""基本查询"""

ret = user.first()  # 取第一条数据，可使用 ret.字段名 取值
user.all()  # 取所有数据


"""条件查询 filter()"""

user.filter(User.id >= 2, User.name == 'user03').all()
# 查询 id >= 2 且 name == 'user03' 的所有数据


"""传参查询 filter_by()"""

user.filter_by(id=3, name='user03').first()
# 查询 id=3 且 name='user03' 的第一条数据


"""更多查询操作"""

# 查 name='user01' 或 age='20' 的数据
db_session.query(User).filter(or_(User.name == 'user01', User.age == '20')).all()

# 筛选字段查询
db_session.query(User.name, User.gender).filter(User.gender == '女').all()

# 重新指定字段名
db_session.query(User.name.label('n')).filter(User.name == 'user01').all()

# 对查询结果进行排序
db_session.query(User).order_by(User.id.desc())
# order_by(User.id.desc())：表示根据id字段进行倒序(desc)，可用asc来正序

# 查id为 1-3 的数据，并根据进行倒序
db_session.query(User).filter(User.id.between(1, 3)).order_by(User.id.desc())

# in_，查id为 1 or3 or 6 的数据
db_session.query(User).filter(User.id.in_([1, 3, 6]))
# 还可用 ~ or notin_ 进行取反，例如：~User.id.in_([1, 3, 6] or User.id.notin_([1, 3, 6]

# 指定值查询，查 id < 3 且 name = 'user02' 的数据
db_session.query(User).filter(text('id<:value and name=:name')).params(value=3, name='user02')

# 执行原声sql，查 name = 'user09' 的数据
db_session.query(User).from_statement(text('select * from user where name=:name').params(name='user09'))

# 分组查询，根据性别分组后统计男女个数
db_session.query(func.count(User.gender).label('count'), User.gender).group_by(User.gender)
# 更多方法：func.max() func.min() func.sum()

# 通配符，查 name 结尾为 '4' 的数据
db_session.query(User).filter(User.name.like('%4'))

# 限制查询，查索引为[0:3]的数据（顾头不顾尾）
db_session.query(User)[0:3]


# 关闭会话对象
db_session.close()

更新数据 

# 用于创建sessionmaker会话对象
from sqlalchemy.orm import sessionmaker

# 导入我们创建的User类与数据库引擎engine
from blog.create_table import User, engine

# 创建sessionmaker会话对象，将数据库引擎engine交给sessionmaker
Session = sessionmaker(engine)

# 打开会话对象Session
db_session = Session()


# 将 id == 1 的数据的 name 更新为 '路招摇'
db_session.query(User).filter(User.id == 1).update({'name': '路招摇'})
# 返回受影响的行数

# 提交
db_session.commit()


# 将所有的数据更新为 '路招摇'
db_session.query(User).filter(User.id > 0).update({'name':  '路招摇'})

# 提交
db_session.commit()


"""更多更新操作"""

# 在原有值上增加，将所有的数据的name后面加'S'
db_session.query(User).filter(User.id > 0).update({User.name: User.name + 'S'}, synchronize_session=False)

# 在原有值上增加，将所有的数据的age加1
db_session.query(User).filter(User.id > 0).update({'age': User.age + 1}, synchronize_session='evaluate')


# 提交
db_session.commit()

# 关闭回话
db_session.close()

删除数据 

# 用于创建sessionmaker会话对象
from sqlalchemy.orm import sessionmaker

# 导入我们创建的User类与数据库引擎engine
from blog.create_table import User, engine

# 创建sessionmaker会话对象，将数据库引擎engine交给sessionmaker
Session = sessionmaker(engine)

# 打开会话对象Session
db_session = Session()


# DELETE FROM 'user' WHERE id=1;
db_session.query(User).filter(User.id == 1).delete()
# 返回受影响的行数


# 提交
db_session.commit()

# 关闭回话
db_session.close()

一对多操作 ForeignKey

创建数据表及关系 relationship 

from sqlalchemy.ext.declarative import declarative_base  # 导入官宣模型
from sqlalchemy import create_engine  # 用于创建数据库引擎
from sqlalchemy import Column  # 用于创建字段
from sqlalchemy import INT, Integer, INTEGER  # 这3个是一样的，使用其中一种即可
from sqlalchemy import String, VARCHAR  # varchar类型，使用其中一个即可
from sqlalchemy import ForeignKey  # 用于创建外键关系
from sqlalchemy.orm import relationship  # 用于创建爱对象关系

# 实例化官宣模型，Base就是ORM模型
Base = declarative_base()


class Person(Base):
    __tablename__ = 'person'
    id = Column(Integer, primary_key=True)  #  默认自增：autoincrement=True
    name = Column(VARCHAR(32))

    # 创建外键关系
    team_id = Column(INT, ForeignKey('team.id'))
    # 'team.id'：表名.字段名

    # 创建对象关系
    p2t = relationship('Team', backref='t2p')
    # 'Team'：类名
    # backref='t2p'：反向查询字段名称


class Team(Base):
    __tablename__ = 'team'
    id = Column(Integer, primary_key=True)
    name = Column(VARCHAR(32))


# 创建数据库引擎
engine = create_engine('mysql+pymysql://zyk:[email protected]@localhost:3306/db01?charset=utf8')

# 将继承Base的所有的Class序列化成数据表
Base.metadata.create_all(engine)

基于 relationship 增加数据 

# 用于创建sessionmaker会话对象
from sqlalchemy.orm import sessionmaker

# 导入我们创建的类与数据库引擎
from blog.一对多.create_table_ForeignKey import Person, Team, engine

# 创建sessionmaker会话对象，将数据库引擎engine交给sessionmaker
Session = sessionmaker(engine)

# 打开会话对象Session
db_session = Session()


"""正向添加"""

# per_obj = Person(name='per01', p2t=Team(name='开发'))
db_session.add(per_obj)  # 将创建的所有数据添加到会话对象中
db_session.commit()  # 提交


"""反向添加"""

tem_obj = Team(name='运维')
tem_obj.t2p = [
    Person(name='per02'),
    Person(name='per03'),
]
db_session.add(tem_obj)  # 将创建的所有数据添加到会话对象中
db_session.commit()  # 提交


# 关闭回话
db_session.close()

基于 relationship 查询数据 

# 用于创建sessionmaker会话对象
from sqlalchemy.orm import sessionmaker

# 导入我们创建的类与数据库引擎
from blog.一对多.create_table_ForeignKey import Person, Team, engine

# 创建sessionmaker会话对象，将数据库引擎engine交给sessionmaker
Session = sessionmaker(engine)

# 打开会话对象Session
db_session = Session()


"""正向查询"""

stu_obj = db_session.query(Person).filter(Person.name == 'per01').first()
print(stu_obj.name, stu_obj.p2t.name)


"""反向查询"""

tem_list = db_session.query(Team).all()
for tem in tem_list:
    for per in tem.t2p:
        print(tem.name, per.name)


# 关闭回话
db_session.close()

多对多操作 ManyToMany

创建数据表及关系 relationship 

from sqlalchemy.ext.declarative import declarative_base  # 导入官宣模型
from sqlalchemy import create_engine  # 用于创建数据库引擎
from sqlalchemy import Column  # 用于创建字段
from sqlalchemy import INT, Integer, INTEGER  # 这3个是一样的，使用其中一种即可
from sqlalchemy import String, VARCHAR  # varchar类型，使用其中一个即可
from sqlalchemy import ForeignKey  # 用于创建外键关系
from sqlalchemy.orm import relationship  # 用于创建爱对象关系

# 实例化官宣模型，Base就是ORM模型
Base = declarative_base()


class Staff(Base):
    __tablename__ = 'staff'
    id = Column(INTEGER, primary_key=True)  #  默认自增：autoincrement=True
    name = Column(String(32))

    s2d = relationship('Dep', backref='d2s', secondary='work')
    # 'Dep'：类名
    # backref='t2p'：反向查询字段名称
    # secondary='work'：指定第三张表


class Dep(Base):
    __tablename__ = 'dep'
    id = Column(INTEGER, primary_key=True)
    name = Column(String(32))


class Work(Base):
    __tablename__ = 'work'
    id = Column(INTEGER, primary_key=True)
    staff_id = Column(INTEGER, ForeignKey('staff.id'))
    dep_id = Column(INTEGER, ForeignKey('dep.id'))
    # 'dep.id'：表名.字段名


# 创建数据库引擎
engine = create_engine('mysql+pymysql://zyk:[email protected]@localhost:3306/db01?charset=utf8')

# 将继承Base的所有的Class序列化成数据表
Base.metadata.create_all(engine)

基于 relationship 增加数据 

# 用于创建sessionmaker会话对象
from sqlalchemy.orm import sessionmaker

# 导入我们创建的类与数据库引擎
from blog.多对多.create_table_ManyToMany import Staff, Dep, Work, engine

# 创建sessionmaker会话对象，将数据库引擎engine交给sessionmaker
Session = sessionmaker(engine)

# 打开会话对象Session
db_session = Session()


"""正向添加"""

staff_obj = Staff(name='staff01')
staff_obj.s2d = [Dep(name='开发')]  # 注意：# 多对多创建使用列表
db_session.add(staff_obj)  # 将创建的所有数据添加到会话对象中
db_session.commit()  # 提交


"""反向添加"""

dep_obj = Dep(name='运维')
dep_obj.d2s = [
    Staff(name='staff02'),
    Staff(name='staff03'),
]
db_session.add(dep_obj)  # 将创建的所有数据添加到会话对象中
db_session.commit()  # 提交


# 关闭回话
db_session.close()

基于 relationship 查询数据 

# 用于创建sessionmaker会话对象
from sqlalchemy.orm import sessionmaker

# 导入我们创建的类与数据库引擎
from blog.多对多.create_table_ManyToMany import Staff, Dep, Work, engine

# 创建sessionmaker会话对象，将数据库引擎engine交给sessionmaker
Session = sessionmaker(engine)

# 打开会话对象Session
db_session = Session()


"""正向查询"""

staff_list = db_session.query(Staff).all()
for staff in staff_list:
    for dep in staff.s2d:
        print(staff.name, dep.name)


"""反向查询"""

dep_list = db_session.query(Dep).all()
for dep in dep_list:
    for staff in dep.d2s:
        print(dep.name, staff.name)


# 关闭回话
db_session.close()

上一篇： 【树状数组】普通、二维、离线树状数组的(单点修改，单点查询，区间修改，区间查询)模板及应用总结

下一篇： 我要打可乐

推荐阅读