【Python SQLAlchemy】 𢓞

原文: http://blog.gqylpy.com/gqy/376

"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:root@zyk@localhost:3306/db01?charset=utf8')
# mysql+pymysql：使用的是pymysql
# root:root@zyk：用户名:密码
# 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:blog@zyk@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:blog@zyk@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()

"

原文: http://blog.gqylpy.com/gqy/376

原文地址：https://www.cnblogs.com/gqy02/p/11374964.html