介绍

SQLAlchemy 是一个 ORM 框架，可以帮助我们使用面向对象的方式快速实现数据库操作。

组成部分：

Engine，框架的引擎；
Connection Pooling，数据库连接池；
Dialect，选择连接数据库的 DB API 种类；
Schema/Types，架构和类型；
SQL Exprression Language，SQL 表达式语言；

SQLAlchemy 本身无法操作数据库，其必须依赖 pymysql 等第三方插件、Dialect（方言）用于和数据 API 进行交流，根据配置文件的不同调用不同的数据库 API ，从而实现对数据库的操作，如：

MySQLDB
    mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
    
pymysql
    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
    
MySQL-Connector
    mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
    
cx_Oracle
    oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
    
更多：http://docs.sqlalchemy.org/en/latest/dialects/index.html

使用

安装

pip3 install sqlalchemy

创建和删除表

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index

Base = declarative_base()


class Users(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    age = Column(Integer, nullable=False)
    name = Column(String(32), index=True, nullable=False)


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


def init_db():
    # 创建表
    Base.metadata.create_all(engine)


def drop_db():
    Base.metadata.drop_all(engine)


if __name__ == '__main__':
    drop_db()
    init_db()

单表操作

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index

Base = declarative_base()


class Users(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)    age = Column(Integer, nullable=False)
    name = Column(String(32), index=True, nullable=False)


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

# 每次执行数据库操作时，都需要创建一个session
session = Session()

新增操作

# 添加单条
obj = Users(name="zhangsan",age=20)
session.add(obj)
# 添加多条
obj1 = Users(name='lisi',age=12)
obj2 = Users(name='wangwu',age=14)
session.add_all([obj1, obj2])
# 提交事务
session.commit()

单表查询

# 查询所有
user_query = session.query(Users)
print(user_query)  # SELECT users.id AS users_id, users.name AS users_name FROM users
print(user_query.all())  # [<__main__.Users object at 0x0000000003A3E2B0>, <__main__.Users object at 0x0000000003A3E320>, <__main__.Users object at 0x0000000003A3E390>]

# 过滤条件
user_query = user_query.filter(Users.id > 1)
print(user_query)  # SELECT users.id AS users_id, users.name AS users_name FROM users WHERE users.id > %(id_1)s
print(user_query.all())  # [<__main__.Users object at 0x000000000395E4A8>, <__main__.Users object at 0x000000000395E518>]

# 取第一个
print(user_query.first())  # <__main__.Users object at 0x000000000392DC88>

删除操作

# 删除 返回删除条数
print(session.query(Users).filter(Users.id > 1).delete())  # 2
session.commit()

修改操作

# 修改方式一
user_obj = session.query(Users).filter(Users.id == 1).first()
user_obj.name = '张大三'
session.commit()

# 修改方式二
session.query(Users).filter(Users.id == 1).update({Users.name: '张小三'})
session.commit()

# 在原来值基础修改
session.query(Users).filter(Users.id == 1).update({Users.name: Users.name + '1'}, synchronize_session=False)
session.query(Users).filter(Users.id == 1).update({'age': Users.age + 2}, synchronize_session='evaluate')
session.commit()

其它操作

# 通过字段过滤
session.query(Users).filter_by(name='zhangsan').all()
# filter 中过滤多条件时它们的关系是 and
session.query(Users).filter(Users.id > 1, Users.name == 'lisi').all()
# between
session.query(Users).filter(Users.id.between(1,2)).all()
# in
session.query(Users).filter(Users.id.in_([1, 3])).all()
# not in
session.query(Users).filter(~Users.id.in_([1, 3])).all()
# 子查询
session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='zhangsan'))).all()
# and 和 or 操作
from sqlalchemy import and_, or_
session.query(Users).filter(and_(Users.id > 3, Users.name == 'lisi')).all()
session.query(Users).filter(or_(Users.id < 2, Users.name == 'zhangsan')).all()
session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == 'zhangsan', Users.id > 3),
    )).all()
# like
session.query(Users).filter(Users.name.like('z%')).all()
# not like
session.query(Users).filter(~Users.name.like('z%')).all()
# 切片 含首不含尾
session.query(Users)[0:1]
# order by
session.query(Users).order_by(Users.name).all()
# order by desc
session.query(Users).order_by(Users.name.desc(),Users.id.asc()).all()
# group by & having & max & sum & min & avg
from sqlalchemy.sql import func
session.query(func.max(Users.age), func.sum(Users.age), func.min(Users.age), func.avg(Users.age)).group_by(Users.age).having(func.min(Users.age) > 14).all()  # [(26, Decimal('26'), 26, Decimal('26.0000'))]
# 查询指定列 & as
ret = session.query(Users.name,Users.age,Users.name.label('nick_name')).all()
[print(user_item[0],user_item.age,user_item.nick_name) for user_item in ret]
# 连表
# where xx.id = xx.id
session.query(Users, Favor).filter(Users.id == Favor.nid).all()
# inner join
session.query(Person).join(Favor).all()
# outer join
session.query(Person).join(Favor, isouter=True).all()
# 组合
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
# 合并去重
ret = q1.union(q2).all()
# 合并不去重
ret = q1.union_all(q2).all()

一对多操作

创建外键关系

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index

Base = declarative_base()


class Depart(Base):
    __tablename__ = 'depart'
    id = Column(Integer, primary_key=True)
    title = Column(String(32), nullable=True)


class Users(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)
    dept_id = Column(Integer, ForeignKey('depart.id'))


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


def init_db():
    # 创建表
    Base.metadata.create_all(engine)


def drop_db():
    Base.metadata.drop_all(engine)


if __name__ == '__main__':
    drop_db()
    init_db()

初始化 Session

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index

Base = declarative_base()


class Depart(Base):
    __tablename__ = 'depart'
    id = Column(Integer, primary_key=True)
    title = Column(String(32), nullable=True)


class Users(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)
    dept_id = Column(Integer, ForeignKey('depart.id'))


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

# 每次执行数据库操作时，都需要创建一个session
session = Session()

联表查询

# 查询所有用户名称 & 所属部门名称
user_list = session.query(Users, Depart).join(Depart, Users.dept_id == Depart.id).all()
print(user_list)
'''
[(<__main__.Users object at 0x000000000395C8D0>, <__main__.Depart object at 0x000000000395C940>),
(<__main__.Users object at 0x000000000395C9B0>, <__main__.Depart object at 0x000000000395C940>),
(<__main__.Users object at 0x000000000395CA20>, <__main__.Depart object at 0x000000000395CA90>)]
'''
[print(user[0].name, user[1].title) for user in user_list]
'''
张三 后勤部
李四 后勤部
王五 财务部
'''
#  SELECT users.name AS users_name, depart.title AS depart_title FROM users INNER JOIN depart ON users.dept_id = depart.id
user_list = session.query(Users.name, Depart.title).join(Depart, Users.dept_id == Depart.id).all()
print(user_list, type(user_list))  # [('张三', '后勤部'), ('李四', '后勤部'), ('王五', '财务部')]
[print(user.name, user.title) for user in user_list]
'''
张三 后勤部
李四 后勤部
王五 财务部
'''
#  SELECT users.name AS users_name, depart.title AS depart_title FROM users LEFT OUTER JOIN depart ON users.dept_id = depart.id
user_list = session.query(Users.name, Depart.title).join(Depart, Users.dept_id == Depart.id, isouter=True)
print(user_list)

relationship 使用

# 正向查询 ：查询所有用户关联的部门名称
user_list = session.query(Users).all()
[print(user.name, user.depart, user.depart.title) for user in user_list]
'''
张三 <__main__.Depart object at 0x00000000039697F0> 后勤部
李四 <__main__.Depart object at 0x00000000039697F0> 后勤部
王五 <__main__.Depart object at 0x0000000003969898> 财务部
'''
# 反向查询 ：查询后勤部所有用户名称
depart_obj = session.query(Depart).filter_by(id=1).first()
[print(user.name, user) for user in depart_obj.users]
'''
张三 <__main__.Users object at 0x000000000395FA20>
李四 <__main__.Users object at 0x000000000395F978>
'''

# 关联新增 ：创建一个部门'酱油部'，再在该部门中添加一个员工'李德刚'
# 方式一：
user_obj = Users(name='李德刚')
depart_obj = Depart(title='酱油部', users=[user_obj])
session.add(depart_obj)
session.commit()
# 方式二：
depart_obj = Depart(title='酱油部')
user_obj = Users(name='李德刚', depart=depart_obj)
session.add(user_obj)
session.commit()

多对多操作

创建多对多关系

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index

Base = declarative_base()


class Permission(Base):
    '''权限'''
    __tablename__ = 'permission'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)


class User(Base):
    '''用户'''
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)


class Permission2User(Base):
    __tablename__ = 'permission2user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    user_id = Column(Integer, ForeignKey('user.id'))
    permission_id = Column(Integer, ForeignKey('permission.id'))

    __table_args__ = (
        # 联合唯一索引
        UniqueConstraint('user_id', 'permission_id', name='uc_user_permission'),
    )


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


def init_db():
    # 创建表
    Base.metadata.create_all(engine)


def drop_db():
    Base.metadata.drop_all(engine)


if __name__ == '__main__':
    drop_db()
    init_db()

初始化 session

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 sessionmaker, relationship

Base = declarative_base()


class Permission(Base):
    '''权限'''
    __tablename__ = 'permission'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)
    # 第一个参数：指定关联的模型类，secondary ：指定通过该表创建多对多关系，backref ：反向查询时字段
    user_list = relationship('User', secondary='permission2user', backref='permission_list')


class User(Base):
    '''用户'''
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)


class Permission2User(Base):
    __tablename__ = 'permission2user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    user_id = Column(Integer, ForeignKey('user.id'))
    permission_id = Column(Integer, ForeignKey('permission.id'))
    __table_args__ = (
        # 联合唯一索引
        UniqueConstraint('user_id', 'permission_id', name='uc_user_permission'),
    )


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

session = Session()

初始化数据

# 初始化数据
session.add_all(
    [
        Permission(name='新增'),
        Permission(name='查询'),
        User(name='张三'),
        User(name='吴刚')
    ]
)
session.commit()

session.add_all([
    Permission2User(user_id=1,permission_id=1),
    Permission2User(user_id=1,permission_id=2),
    Permission2User(user_id=2,permission_id=2),
])
session.commit()

常用操作示例

list = session.query(Permission2User.id, User.name, Permission.name) \
    .join(Permission, Permission2User.permission_id == Permission.id) \
    .join(User, Permission2User.user_id == User.id).all()
print(list)  # [(3, '吴刚', '查询'), (1, '张三', '新增'), (2, '张三', '查询')]
# 查询拥有新增权限的用户
permission_obj = session.query(Permission).filter_by(name='新增').first()
[print(user_obj.name) for user_obj in permission_obj.user_list]  # 张三
# 查询吴刚拥有的权限
user_obj = session.query(User).filter_by(name='吴刚').first()
[print(permission_obj.name) for permission_obj in user_obj.permission_list]  # 查询

#  创建一个权限，再创建两个用户，让这两个用户拥有这个权限
permission_obj = Permission(name='修改')
permission_obj.user_list = [User(name='王五'), User(name='赵柳')]
session.add(permission_obj)
session.commit()

两种连接方式

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 sessionmaker, relationship, scoped_session

Base = declarative_base()


class User(Base):
    '''用户'''
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)


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


def task1():
    # 去连接池获取一个连接。
    session = SessionFactory()
    ret = session.query(User).all()
    # 将连接交还给连接池。
    session.close()


session = scoped_session(session_factory=SessionFactory)


def task2():
    # 当真正使用 session 时，才从连接池中取一个连接放在 threading.Local 对象的当前线程域中，实现多线程访问时的 session 隔离。
    ret = session.query(User).all()
    # 将连接交还给连接池
    session.remove()


from threading import Thread

for i in range(20):
    t1 = Thread(target=task1)
    t1.start()
    t2 = Thread(target=task2)
    t2.start()

执行原生 SQL

# 添加
cursor = session.execute('insert into user(name) values(:name)', params={"name": '刘能'})
session.commit()
print(cursor.lastrowid)  # 5

# 查询
cursor = session.execute('select * from user')
result = cursor.fetchall()
print(result)  # [(5, '刘能'), (2, '吴刚'), (1, '张三'), (3, '王五'), (4, '赵柳')]

目录CONTENT

SQLAlchemy框架的简单使用

介绍

使用

安装

创建和删除表

单表操作

新增操作

单表查询

删除操作

修改操作

其它操作

一对多操作

创建外键关系

初始化 Session

联表查询

relationship 使用

多对多操作

创建多对多关系

初始化 session

初始化数据

常用操作示例

两种连接方式

执行原生 SQL

评论区