时间戳

首页 > 技术文章 > SQLAlchemy（2）：多表操作 & 连接方式及原生SQL

SQLAlchemy（2）：多表操作 & 连接方式及原生SQL

neozheng 2019-01-24 00:17 原文

一对多：ForeignKey

multitb_models.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()  # Base 要自己实例化

class Depart(Base):
    __tablename__ = "depart"  # 表名
    id = Column(Integer, primary_key=True)
    title = Column(String(32), index=True, nullable=False)

class Users(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)
    depart_id = Column(Integer, ForeignKey("depart.id"))  # ForeignKey 需要导入；"depart.id" ：表名.id （是表名，而不是类名）

    # 与生成表结构无关，仅用于 跨表 查询方便（即不会在 users 这张表中生成 dp 这个字段）
    dp = relationship("Depart", backref='pers')  # relationship() 中的 "Depart" 是类名； dp 是与 Depart 这个类做关联； backref="pers"用于反向查询（由 Depart 查询 Users）


def init_db():
    """
    根据类创建数据库的表
    :return:
    """
    engine = create_engine(  # 创建数据库连接
        "mysql+pymysql://root:tj037778@127.0.0.1:3306/dbtest?charset=utf8",  # mysql 表示要连接的数据库；pymysql 表示用 pymysql 来连接；用户名是 root，密码是123，连接本地的 dbtest 这个数据库
        max_overflow=0,  # 超过连接池大小外最多创建的连接；即超过 pool_size 后最多能溢出多少个连接
        pool_size=5,  # 连接池大小
        pool_timeout=10,  # 池中没有线程（连接）最多等待的时间，否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）；-1表示不重建
    )

    Base.metadata.create_all(engine)  # Base.metadata.create_all() ： 找到当前 py 文件下面 继承了Base的所有的类，在数据库中生成一张表


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

    Base.metadata.drop_all(engine)  # Base.metadata.drop_all() ： 找到当前 py 文件下面 继承了Base的所有的类，在数据库中删除相应的表


if __name__ == "__main__":
    init_db()
    # drop_db()

multitb_crud.py

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from multitb_models import Users,Depart

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

session = Session()

# ############### ForeignKey ##############
# 1. 查询所有的用户 + 所属部门名称
ret1 = session.query(Users,Depart).join(Depart).all()  # .join() 时 默认的 on 是根据 ForeignKey("depart.id")，进行 on users.depart_id = depart.id （默认是通过 ForeignKey 进行连表）
for row in ret1: # 此时 row 为 一个元组，里面的元素为 Users 和 Depart 的对象
    print(row[0].name,row[1].title)
"""
ret1 = session.query(Users,Depart).join(Depart).all() 中的 .join() 也可以指定 on ，如下：
ret1 = session.query(Users,Depart).join(Depart,Users.depart_id == Depart.id).all()
"""
ret2 = session.query(Users.id,Users.name,Depart.title).join(Depart,Users.depart_id==Depart.id).all()  # 只取 users 的 id name 和 depart 的 title
for row in ret2:  # row 也是元组的形式
    print(row.id,row.name,row.title)
"""
.join() 默认是 inner join，想要变成 left join 可以在 join()中设置 isouter=True ，如下：
session.query(Users.id,Users.name,Depart.title).join(Depart,Users.depart_id==Depart.id,isouter=True).all()
另外，SQLAlchemy 的 join() 没有 right join，想要 right join 可以在 query() 中将 Users 和 Depart 调换下位置
注： .join() 后面可以继续 .join() ， 即可以 连很多张表
"""

# 2. relationship 字段：查询所有的用户 + 所属部门名称 （类似 正向查询）
ret3 =  session.query(Users).all()
for row in ret3:
    print(row.id,row.name,row.dp.title)  # row.dp 是 该row对象（Users对象） 对应的 depart这张表中 所关联的外键 对象记录，所以 row.dp.title 即为 Depart.title
# 打印结果：
# 2 neo 开发部
# 3 alex 开发部
# 4 egon 市场部
# 5 wu 运维部

# 3. relationship 字段：查询开发部的所有人员 （类似 反向查询）
obj = session.query(Depart).filter(Depart.title=="开发部").first()  # obj 是 title 为 "开发部" 的一个 Depart对象
print(obj.pers)
# 打印结果：列表的形式；列表中的元素其为所关联的 Users对象
# [<multitb_models.Users object at 0x0000018899C76940>, <multitb_models.Users object at 0x0000018899C76A20>]
for row in obj.pers:
    print(row.id,row.name,obj.title)

# 4. relationship 字段：创建一个名为 “销售部” 的部门，并在该部门中添加一个名为 “maple” 的员工 （一次性创建所有的关联数据）
user1 = Users(name="maple",dp=Depart(title="销售部"))  # 通过这种写法，能在 users 表中创建一个name 为 "maple"的记录，由于其关联的 depart_id 是新添加的，其也能在 depart表中创建一个 title为"销售部"的记录，并自动将 title=="销售部"的depart的记录id添加到 users表中的 depart_id字段（每次都会创建一个 Depart 的实例对象）
session.add(user1)
session.commit()

# 5. relationship 字段：创建一个名为“IT部”的部门，并在该部门中添加多个员工
depart1 = Depart(title="IT部")  # 创建 Depart 的一个对象
depart1.pers = [  # 为 depart1 的反向字段 添加一个列表，列表中是 depart1 所对应的 Users 对象
    Users(name="neo1"),
    Users(name="neo2"),
    Users(name="neo3")
]
session.add(depart1)  # 只需要把 depart1 添加；因为 既有 title，又有 其所对应的 Users 对象
session.commit()


session.close()

 

多对多：m2m

m2m_models.py

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()  # Base 要自己实例化


class Student(Base):
    __tablename__ = "student"
    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)

    # 表中不会额外增加字段，只是为了方便跨表操作
    course_list = relationship("Course", secondary="student2course",
                               backref="student_list")  # 第一个参数表示 和哪张表作关联，第二个参数表示 通过哪张表和 "Course"做关联，第三个参数表示反向字段名


class Course(Base):
    __tablename__ = "course"
    id = Column(Integer, primary_key=True)
    title = Column(String(32), index=True, nullable=False)


# SQLAlchemy 的多对多要自己创建第三张表
class Student2Course(Base):
    __tablename__ = "student2course"
    id = Column(Integer, primary_key=True, autoincrement=True)
    student_id = Column(Integer, ForeignKey("student.id"))
    course_id = Column(Integer, ForeignKey("course.id"))

    # 建立联合唯一
    __table_args__ = (
        UniqueConstraint("student_id", "course_id", name="uix_stu_cou"),  # UniqueConstraint：联合唯一索引；该索引的名字是 uix_stu_cou
        # Index() 可用于 联合索引（没有唯一的要求）
    )

def init_db():
    """
    根据类创建数据库的表
    :return:
    """
    engine = create_engine(  # 创建数据库连接
        "mysql+pymysql://root:tj037778@127.0.0.1:3306/dbtest?charset=utf8",  # mysql 表示要连接的数据库；pymysql 表示用 pymysql 来连接；用户名是 root，密码是123，连接本地的 dbtest 这个数据库
        max_overflow=0,  # 超过连接池大小外最多创建的连接；即超过 pool_size 后最多能溢出多少个连接
        pool_size=5,  # 连接池大小
        pool_timeout=10,  # 池中没有线程（连接）最多等待的时间，否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）；-1表示不重建
    )

    Base.metadata.create_all(engine)  # Base.metadata.create_all() ： 找到当前 py 文件下面 继承了Base的所有的类，在数据库中生成一张表


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

    Base.metadata.drop_all(engine)  # Base.metadata.drop_all() ： 找到当前 py 文件下面 继承了Base的所有的类，在数据库中删除相应的表


if __name__ == "__main__":
    init_db()
    # drop_db()

m2m_crud.py

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from m2m_models import Student, Course, Student2Course

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

session = Session()

# ####################### 多对多：m2m ###########################
# 一、基本用法（不使用 relationship）
# 1. 录入数据
# session.add_all([
#     Student(name="neo"),
#     Student(name="alex"),
#     Course(title="生物"),
#     Course(title="体育"),
# ])
# session.commit()

# 录入 关系表 的数据
# session.add_all([
#     Student2Course(student_id=1,course_id=1),
#     Student2Course(student_id=1,course_id=2),
#     Student2Course(student_id=2,course_id=1),
# ])
# session.commit()

# 2. 三张表关联：查询每个学生对应的课程
ret1 = session.query(Student2Course.id,Student.name,Course.title).join(Student,Student2Course.student_id==Student.id,isouter=True).join(Course,Student2Course.course_id==Course.id,isouter=True).order_by(Student2Course.id.asc())
for row in ret1:
    print(row)
# 打印结果：
# (2, 'neo', '生物')
# (3, 'neo', '体育')
# (4, 'alex', '生物')

# 3. 查询 neo 对应的所有课
ret1 = session.query(Student2Course.id,Student.name,Course.title).join(Student,Student2Course.student_id==Student.id,isouter=True).join(Course,Student2Course.course_id==Course.id,isouter=True).filter(Student.name=="neo").order_by(Student2Course.id.asc()).all()
for row in ret1:
    print(row)

# 二、 使用 relationship
# 4. 查询 neo 对应的所有课 （正向查询）
obj = session.query(Student).filter(Student.name=="neo").first()  # name=="neo" 的 Student 的对象
for item in obj.course_list: # item 是 obj.course_list 这个列表中的一个个对象
    print(item.id,item.title)

# 5. 查询选了“生物”的所有的人 （反向查询）
course_obj = session.query(Course).filter(Course.title=="生物").first()
for item in course_obj.student_list:  # course_obj.student_list
    print(item.id,item.name)

# 6. 创建一个新的课程“英语”，并创建两个学生并让这两个学生对应新创建的“英语”课程
new_course_obj = Course(title="英语")
new_course_obj.student_list = [  # 内部会自动创建关系
    Student(name="alina"),
    Student(name="mike")
]
session.add(new_course_obj)
session.commit()

session.close()

 

SQLAlchemy 的两种连接方式

方式一：将获取连接的操作放到线程函数里面

import time
import threading

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from db import Users

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


def task(arg):
    session = Session()  # 多线程的情况下，获取数据库连接的操作要放到线程函数里面，而不能放到全局

    obj1 = Users(name="alex1")
    session.add(obj1)

    session.commit()
    session.close()  # 将连接交还给连接池


for i in range(10):
    t = threading.Thread(target=task, args=(i,))
    t.start()

方式二：利用 scoped_session （推荐使用这种：写法简单）

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
from models import Users

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

session = scoped_session(Session)  # scoped_session() 的原理是 threading.local()：为每个线程获取一个连接

def task(arg):

    obj1 = Users(name="alex1")
    session.add(obj1)

    session.commit()
    session.remove()  # session.remove() ： 将连接交还给连接池

import threading
for i in range(10):
    t = threading.Thread(target=task, args=(i,))
    t.start()

 

SQLAlchemy 执行原生SQL

方式一：

import time
import threading

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from db import Users, Hosts

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

session = Session()

# 查询
# cursor = session.execute('select * from users')  # 和 pymysql 的用法一样
# result = cursor.fetchall()

# 添加
cursor = session.execute('insert into users(name) values(:value)',params={"value":'wupeiqi'}) # 此处的 字符串格式化 不是利用 "%"，而是 :value
session.commit()
print(cursor.lastrowid)

session.close()

方式二：

import time
import threading
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engine
 
engine = create_engine(
    "mysql+pymysql://root:123@127.0.0.1:3306/t1?charset=utf8",
    max_overflow=0,  # 超过连接池大小外最多创建的连接
    pool_size=5,  # 连接池大小
    pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错
    pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
 
 
def task(arg):
    conn = engine.raw_connection()  # 获取连接 
    cursor = conn.cursor()  # 创建 游标
    cursor.execute(
        "select * from t1"
    )
    result = cursor.fetchall()
    cursor.close()
    conn.close()
 
 
for i in range(20):
    t = threading.Thread(target=task, args=(i,))
    t.start()

 

推荐阅读