Flask框架-SqlAlchemy玩转MySQL

Flask框架-SqlAlchemy玩转MySQL 一. ORM框架

1. Web 开发中，一个重要的组成部分便是数据库了。Web 程序中最常用的莫过于关系型数据库了，也称 SQL 数据库。另外，文档数据库（如 mongodb）、键值对数据库（如 redis）近几年也逐渐在 web 开发中流行起来，我们习惯把这两种数据库称为 NoSQL 数据库。
2. 大多数的关系型数据库引擎（比如 MySQL、Postgres 和 SQLite）都有对应的 Python 包。在这里，我们不直接使用这些数据库引擎提供的 Python 包，而是使用对象关系映射（Object-Relational Mapper, ORM）框架，它将低层的数据库 *** 作指令抽象成高层的面向对象 *** 作。也就是说，如果我们直接使用数据库引擎，我们就要写 SQL *** 作语句，但是，如果我们使用了 ORM 框架，我们对诸如表、文档此类的数据库实体就可以简化成对 Python 对象的 *** 作。
3. Python 中最广泛使用的 ORM 框架是 SQLAlchemy，它是一个很强大的关系型数据库框架，不仅支持高层的 ORM，也支持使用低层的 SQL *** 作，另外，它也支持多种数据库引擎，如 MySQL、Postgres 和 SQLite 等。

二. 介绍

SQLAlchemy是一个基于Python实现的ORM框架。该框架建立在 DB API之上，使用关系对象映射进行数据库 *** 作，简言之：将类和对象转换成SQL，然后使用数据库模块调用DB-API执行SQL并获取执行结果。

pip install sqlalchemy

由于sqlalchemy模块依赖于第三方DB-API模块,因此,我们还需要安装数据库模块

pip install pymysql

三. 框架基本结构

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-nKuAu4MR-1641009641116)(//upload-images.jianshu.io/upload_images/3269979-45ea4491dbec8589.png?imageMogr2/auto-orient/strip|imageView2/2/w/1076/format/webp)]
 
 
 
组成部分：

Engine，框架的引擎

Connection Pooling ，数据库连接池

Dialect，选择连接数据库的DB API种类

Schema/Types，架构和类型

SQL Exprression Language，SQL表达式语言

四. 数据库连接串 数据库引擎URLMySQL-Pythomysql+mysqldb://:@[:]/pymysqlmysql+pymysql://:@/[?]MySQL-Connectormysql+mysqlconnector://:@[:]/Postgrespostgresql://username:password@hostname/databaseSQLite (Unix)sqlite:absolute/path/to/databaseSQLite (Windows)sqlite:///c:/absolute/path/to/databasecx_Oracleoracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value…]

SQLAlchemy本身无法 *** 作数据库，其必须以来pymsql等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的 *** 作，下面的表格中，username 和 password 表示登录数据库的用户名和密码，hostname 表示 SQL 服务所在的主机，可以是本地主机（localhost）也可以是远程服务器，database 表示要使用的数据库。有一点需要注意的是，SQLite 数据库不需要使用服务器，它使用硬盘上的文件名作为 database。如：

数据库引擎URLMySQL-Pythomysql+mysqldb://:@[:]/pymysqlmysql+pymysql://:@/[?]MySQL-Connectormysql+mysqlconnector://:@[:]/Postgrespostgresql://username:password@hostname/databaseSQLite (Unix)sqlite:absolute/path/to/databaseSQLite (Windows)sqlite:///c:/absolute/path/to/databasecx_Oracleoracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value…]

更多

五. 使用 5.1 执行原生SQL

• 方式一

import time
import threading
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engine

engine = create_engine(
    "mysql+pymysql://zff:[email protected]:3306/zff?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 USER "
    )
    result = cursor.fetchall()
    cursor.close()
    conn.close()
    print(result)


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

• 方式二

import threading
from sqlalchemy import create_engine

engine = create_engine("mysql+pymysql://zff:[email protected]:3306/zff", max_overflow=0, pool_size=5)


def task(arg):
    conn = engine.contextual_connect()
    with conn:
        cur = conn.execute(
            "select * from USER "
        )
        result = cur.fetchall()
        print(result)


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

• 方式三

import threading
from sqlalchemy import create_engine

engine = create_engine("mysql+pymysql://zff:[email protected]:3306/zff", max_overflow=0, pool_size=5)


def task(arg):
    cur = engine.execute("select * from USER ")
    # cur = engine.execute("select sleep(10)")
    result = cur.fetchall()
    cur.close()
    print(result)


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

⚠️ 查看连接 show status like ‘Threads%’;

六. ORM使用 6.1 创建数据库表

• 单表示例

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

# 基类
base = declarative_base()


class Users(base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)
    email = Column(String(32), unique=True)
    ctime = Column(DateTime, default=datetime.datetime.now)
    # extra = Column(Text, nullable=True)

    __table_args__ = (
        # UniqueConstraint('id', 'name', name='uix_id_name'),   # 联合唯一索引
        # Index('ix_id_name', 'name', 'email'),                 # 联合索引
    )


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

    base.metadata.create_all(engine)


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

    base.metadata.drop_all(engine)


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

• FK/M2M示例

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()


# ##################### 单表示例 #########################
class Users(base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True)
    age = Column(Integer, default=18)
    email = Column(String(32), unique=True)
    ctime = Column(DateTime, default=datetime.datetime.now)
    extra = Column(Text, nullable=True)

    __table_args__ = (
        # UniqueConstraint('id', 'name', name='uix_id_name'),
        # Index('ix_id_name', 'name', 'extra'),
    )


class Hosts(base):
    __tablename__ = 'hosts'

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True)
    ctime = Column(DateTime, default=datetime.datetime.now)


# ##################### 一对多示例 #########################
class Hobby(base):
    __tablename__ = 'hobby'
    id = Column(Integer, primary_key=True)
    caption = Column(String(50), default='篮球')


class Person(base):
    __tablename__ = 'person'
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    hobby_id = Column(Integer, ForeignKey("hobby.id"))

    # 与生成表结构无关，仅用于查询方便
    hobby = relationship("Hobby", backref='pers')  # backref 反向查询


# ##################### 多对多示例 #########################

class Server2Group(base):
    __tablename__ = 'server2group'
    id = Column(Integer, primary_key=True, autoincrement=True)  # 自增主键
    server_id = Column(Integer, ForeignKey('server.id'))
    group_id = Column(Integer, ForeignKey('group.id'))


class Group(base):
    __tablename__ = 'group'
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False)

    # 与生成表结构无关，仅用于查询方便
    servers = relationship('Server', secondary='server2group', backref='groups')  # secondary: 指定关系表


class Server(base):
    __tablename__ = 'server'

    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(64), unique=True, nullable=False)


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

    base.metadata.create_all(engine)


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

    base.metadata.drop_all(engine)


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

指定关联列: hobby = relationship(“Hobby”, backref=‘pers’,foreign_keys=“Person.hobby_id”)

6.1 ORM基本使用

• 初步使用

#! /usr/bin/env python
# -*- coding: utf-8 -*-
# __author__ = "shuke"
# Date: 2018/5/14

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import *

engine = create_engine("mysql+pymysql://zff:[email protected]:3306/zff?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)

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

# ############# 执行ORM *** 作 #############
obj1 = Users(name="tome", age=19, email="[email protected]")
session.add(obj1)

# 提交事务
session.commit()
# 关闭session
session.close()

• 原生SQL

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.sql import text
from models import *

engine = create_engine("mysql+pymysql://zff:[email protected]:3306/zff?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)

session = Session()

# 查询
# cursor = session.execute('select * from users')
# result = cursor.fetchall()

# 添加
with engine.connect() as con:
    data = ({'name': 'zhangsan', 'age': 20, 'value': '[email protected]'},
            {'name': 'lisi', 'age': 20, 'value': '[email protected]'})
    statement = text("""insert into users(name,age,email) values(:name,:age,:value)""")
    for line in data:
        cursor = con.execute(statement, **line)

print(cursor.lastrowid)
session.close()

• 基本增删改查

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.sql import text
from models import *

engine = create_engine("mysql+pymysql://zff:[email protected]:3306/zff?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)

session = Session()

# 1. ################ 添加 ################
"""
obj1 = Users(name="jack", age=19, email="[email protected]")
session.add(obj1)

session.add_all([
    Users(name="wang", age=19, email="[email protected]"),
    Users(name="lucy", age=19, email="[email protected]"),
    Hosts(name="jav-pingtai03br-p002.gru1.blue.net"),
])
session.commit()
"""

# 2. ################ 删除 ################
"""
session.query(Users).filter(Users.id > 2).delete()
session.commit()
"""
# 3. ################ 修改 ################
"""
session.query(Users).filter(Users.id > 0).update({"name" : "shuke"})
session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "163"}, synchronize_session=False)
session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate")
session.commit()

# sqlalchemy 利用 session 执行 delete 时有一个 synchronize_session 参数用来说明 session 删除对象时需要执行的策略，共三个选项：
1. False
不同步 session，如果被删除的 objects 已经在 session 中存在，在 session commit 或者 expire_all 之前，这些被删除的对象都存在 session 中。
不同步可能会导致获取被删除 objects 时出错。

# 2. fetch
删除之前从 db 中匹配被删除的对象并保存在 session 中，然后再从 session 中删除，这样做是为了让 session 的对象管理 identity_map 得知被删除的对象究竟是哪些以便更新引用关系。

# 3. evaluate
# 默认值。根据当前的 query criteria 扫描 session 中的 objects，如果不能正确执行则抛出错误，这句话也可以理解为，如果 session 中原本就没有这些被删除的 objects，扫描当然不会发生匹配，相当于匹配未正确执行。
注意这里报错只会在特定 query criteria 时报错，比如 in  *** 作。

session.query(Users).filter(Users.id.in_([1,2,3])).delete()
sqlalchemy.exc.InvalidRequestError: Could not evaluate current criteria in Python. Specify 'fetch' or False for the synchronize_session parameter.
"""

# 3. ################ 查询 ################
"""
r1 = session.query(Users).all()
r2 = session.query(Users.name.label('username'), Users.age).all()   # 别名
r3 = session.query(Users).filter(Users.name == "shuke").all()
r4 = session.query(Users).filter_by(name='shuke').all()
r5 = session.query(Users).filter_by(name='shuke').first()
r6 = session.query(Users).filter(text("id<:value and name=:name")).params(value=2, name='shuke').order_by(Users.id).all()
r7 = session.query(Users).from_statement(text("SELECt * FROM users where name=:name")).params(name='shuke').all()
"""

# filter和filter_by的区别
"""
filter_by用于简单的列名查询，如：
db.users.filter_by(name='Joe')

filter对于上面的代码可以这样写：
db.users.filter(db.users.name == 'Joe')

对于复杂的查询使用filter，如：
db.users.filter(or_(db.users.name == 'Ryan', db.users.country == 'England'))

注意: filter_by使用的是赋值 =, 而filter使用的是判断 ==

另外：查询时使用like这样写： items = session.query.filter(Users.name == current_user, Users.title.like('%' + keyword + '%')).all()
"""
session.close()

• 常用 *** 作

# 条件
ret = session.query(Users).filter_by(name='shuke').all()
ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.in_([1, 3, 4])).all()
ret = session.query(Users).filter(~Users.id.in_([1, 3, 4])).all()
ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()

# 条件 AND / OR
from sqlalchemy import and_, or_

ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
ret = session.query(Users).filter(or_(Users.id < 3, Users.name == 'shuke')).all()
ret = session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == 'eric', Users.id > 3),
        Users.extra != ""
    )).all()

# 通配符
ret = session.query(Users).filter(Users.name.like('e%')).all()
ret = session.query(Users).filter(~Users.name.like('e%')).all()  # 取反

# 限制
ret = session.query(Users)[1:2]  # limit

# 排序
ret = session.query(Users).order_by(Users.name.desc()).all()  # 按降序排列
ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()  # 根据name按降序排列，再根据id升序排列

# 分组
from sqlalchemy.sql import func

ret = session.query(Users).group_by(Users.extra).all()
ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).all()

ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) > 2).all()

# 连表查询
默认是innerjoin
# 直接让两个表联合。这里join默认是innerjoin，这里没有写他们的对应关系，它们在内部自己找。
# 它是怎么找的呢，在创建表的时候，有类型是foreignkey，是根据它来找的。
ret = session.query(Person).join(Favor).all()
相当于sql语句,两个表通过on，来关联

a.  查看sql
sql = session.query(Person).join(Favor)
print(sql)
'''
inner join打印sql，只打印person表所有字段信息，不打印favor表
SELECt person.nid AS person_nid, person.name AS person_name, person.favor_id AS person_favor_id
FROM person JOIN favor ON favor.nid = person.favor_id
'''

b. isouter=True即left join
ret1 = session.query(Person).join(Favor,isouter=True).all()
sql1 = session.query(Person).join(Favor,isouter=True)
print(sql1)
'''
打印sql，只打印person表所有字段的信息，但是没有打印favor表
SELECt person.nid AS person_nid, person.name AS person_name, person.favor_id AS person_favor_id
FROM person LEFT OUTER JOIN favor ON favor.nid = person.favor_id
'''

c. 两张表的信息都打印出来
ret2 = session.query(Person, Favor).join(Favor, isouter=True).all()
print(ret2)
sql2 = session.query(Person, Favor).join(Favor, isouter=True)
print(sql2)
'''
left join,打印结果：打印person和favor两张表的所有字段。
[(<__main__.Person object at 0x0000000003B34FD0>, 1-white), (<__main__.Person object at 0x0000000003B69BE0>, 2-blue),
(<__main__.Person object at 0x0000000003B69C50>, 2-blue)]

left join,打印sql：打印person和favor两张表的所有字段。
SELECt person.nid AS person_nid, person.name AS person_name, person.favor_id AS person_favor_id,
favor.nid AS favor_nid, favor.caption AS favor_caption
FROM person LEFT OUTER JOIN favor ON favor.nid = person.favor_id
'''

d. 联表，只打印某些字段
ret3 = session.query(Person.name,Favor.caption).join(Favor,isouter=True).all()
print(ret3)
sql3 = session.query(Person.name,Favor.caption).join(Favor,isouter=True)
print(sql3)
'''
left join,打印结果：某些指定字段值
[('qiaomei0', 'white'), ('qiaomei1', 'blue'), ('qiaomei2', 'blue')]
left join,打印sql：某些指定字段值
SELECt person.name AS person_name, favor.caption AS favor_caption
FROM person LEFT OUTER JOIN favor ON favor.nid = person.favor_id
'''

# 关联子查询
subqry = session.query(func.count(Server.id).label("sid")).filter(Server.id == Group.id).correlate(Group).as_scalar()
result = session.query(Group.name, subqry)
"""
SELECt `group`.name AS group_name, (SELECT count(server.id) AS sid 
FROM server 
WHERe server.id = `group`.id) AS anon_1 
FROM `group`
"""

# 组合
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()

• 联表查询(一对多,基于relationship *** 作ForeignKey)

1. 正向查询
   多对一，多的那端开始查，也就是foreignkey写在哪里，从哪里查。
   使用上面的方法非常麻烦，我们用更高效的方法。
   只要在表里加上这一句话：

favor = relationship("Favor", backref='pers')

class Person(base):
    __tablename__ = 'person'
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    favor_id = Column(Integer, ForeignKey("favor.nid"))
    # 与生成表结构无关，仅用于查询方便
    favor = relationship("Favor", backref='pers')
 
    # obj代指的是Person表的每一行数据
    # obj.favor代指favor对象，obj.favor.nid就拿到了Person关联的favor对象的id。
    # 所以你不用做联表，它内部帮你做联表。
ret = session.query(Person).all()
for obj in ret:     # 每个obj就是一行数据。
 
    print(obj.nid,obj.name,obj.favor_id,obj.favor,obj.favor.nid,obj.favor.caption)

1. 反向查询
   多对一，从一的那端开始查，也就是从没写foreignkey的表里反查。
   多对一，从一的那端反查。
   Person和Favor是多对一，假如查询喜欢蓝色的所有人。Favor的caption为blue的所有对应的Person

传统方式，反向查询：

ret3 = session.query(Person.name,Favor.caption).join(Favor,isouter=True).filter(Favor.caption == 'blue').all()

Person表里，写了backref=‘pers’,就相当于在favor表里加了个字段pers,实际是不存在的

favor = relationship("Favor", backref='pers')

如下:

class Person(base):
    __tablename__ = 'person'
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    favor_id = Column(Integer, ForeignKey("favor.nid"))
    # 与生成表结构无关，仅用于查询方便
    favor = relationship("Favor", backref='pers')
 
class Favor(base):
    __tablename__ = 'favor'
    nid = Column(Integer, primary_key=True)
    caption = Column(String(50), default='red', unique=True)
    # Person表里写的：backref='pers',相当于在这里加上字段pers。只是用于查询，不会修改表结构。
    # pers = 。。。。。。。。
    def __repr__(self):
        return "%s-%s" %(self.nid, self.caption)

你可以直接通过Favor对象的pers字段找到跟这个颜色关联的所有person,在数据库里没有真实的字段对应的，只是帮你生成sql语句而已。

# 新方式，反向查询
obj = session.query(Favor).filter(Favor.caption=='blue').first()    # 先找到caption为blue的Favor对象
print(obj.nid)
print(obj.caption)
print(obj.pers)
 
'''
打印结果：
2
blue
[<__main__.Person object at 0x0000000003B5BBE0>, <__main__.Person object at 0x0000000003B5BC50>]
'''

Foreignkey(一对多总结)

Foreignkey和relationship要成对写在一个表里。

class Person(base):
    __tablename__ = 'person'
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    favor_id = Column(Integer, ForeignKey("favor.nid"))
    # 与生成表结构无关，仅用于查询方便
    favor = relationship("Favor", backref='pers')

Person对Favor 是多对一的关系，foreignkey加在了多的那端(Person表)

Person对象.favor.favor的字段：叫做正向查找

Favor对象.pers.person的字段：反向查找

• M2M(基于relationship的m2m关系)

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, Hobby, Person, Group, Server, Server2Group

engine = create_engine("mysql+pymysql://root:[email protected]:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
# 添加
"""
session.add_all([
    Server(hostname='c1.com'),
    Server(hostname='c2.com'),
    Group(name='A组'),
    Group(name='B组'),
])
session.commit()

s2g = Server2Group(server_id=1, group_id=1)
session.add(s2g)
session.commit()

gp = Group(name='C组')
gp.servers = [Server(hostname='c3.com'),Server(hostname='c4.com')]
session.add(gp)
session.commit()

ser = Server(hostname='c6.com')
ser.groups = [Group(name='F组'),Group(name='G组')]
session.add(ser)
session.commit()
"""

# 使用relationship正向查询
"""
v = session.query(Group).first()
print(v.name)
print(v.servers)
"""

# 使用relationship反向查询
"""
v = session.query(Server).first()
print(v.hostname)
print(v.groups)
"""
session.close()

• 多线程执行示例

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:[email protected]:3306/s6", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)

def task(arg):
    session = Session()
    obj1 = Users(name="shuke")
    session.add(obj1)
    session.commit()


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

• 基于scoped_session使得线程安全
  基于ThreadLocal实现

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:[email protected]:3306/s6", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)

"""
# 线程安全，基于本地线程实现每个线程用同一个session
# 特殊的：scoped_session中有原来方法的Session中的一下方法：

public_methods = (
    '__contains__', '__iter__', 'add', 'add_all', 'begin', 'begin_nested',
    'close', 'commit', 'connection', 'delete', 'execute', 'expire',
    'expire_all', 'expunge', 'expunge_all', 'flush', 'get_bind',
    'is_modified', 'bulk_save_objects', 'bulk_insert_mappings',
    'bulk_update_mappings',
    'merge', 'query', 'refresh', 'rollback',
    'scalar'
)

"""
session = scoped_session(Session)

# ############# 执行ORM *** 作 #############
obj1 = Users(name="shuke")
session.add(obj1)

# 提交事务
session.commit()
# 关闭session
session.close()

参考资料:

Flask-SQLAlchemy-武沛齐-博客园

mysql和SQLAlchemy-博客园

使用flask-sqlalchemy玩转MySQL | Wing’s Tech Space

Flask-Migrate的使用 | Wing’s Tech Space