SqlAlchemy玩转MySQL

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并获取执行结果。

1
pip install sqlalchemy

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

1
pip install pymysql

三. 框架基本结构

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

四. 连接串

SQLAlchemy本身无法操作数据库,其必须以来pymsql等第三方插件,Dialect用于和数据API进行交流,根据配置文件的不同调用不同的数据库API,从而实现对数据库的操作,如:
|数据库引擎|URL|
|—-|—-|
|MySQL-Pytho|mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>|
|pymysql| mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]|
|MySQL-Connector|mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>|
|Postgres|postgresql://username:password@hostname/database|
|SQLite (Unix)|sqlite:////absolute/path/to/database|
|SQLite (Windows)|sqlite:///c:/absolute/path/to/database|
|cx_Oracle|oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value…]|
上面的表格中,username 和 password 表示登录数据库的用户名和密码,hostname 表示 SQL 服务所在的主机,可以是本地主机(localhost)也可以是远程服务器,database 表示要使用的数据库。有一点需要注意的是,SQLite 数据库不需要使用服务器,它使用硬盘上的文件名作为 database。
更多

五. 使用

5.1 执行原生SQL

  • 方式一
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    import time
    import threading
    import sqlalchemy
    from sqlalchemy import create_engine
    from sqlalchemy.engine.base import Engine

    engine = create_engine(
    "mysql+pymysql://zff:zff123@127.0.0.1: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()
  • 方式二
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    import threading
    from sqlalchemy import create_engine

    engine = create_engine("mysql+pymysql://zff:zff123@127.0.0.1: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()
  • 方式三
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    import threading
    from sqlalchemy import create_engine

    engine = create_engine("mysql+pymysql://zff:zff123@127.0.0.1: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 创建数据库表

  • 单表示例

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    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:zff123@127.0.0.1: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:zff123@127.0.0.1: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示例

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    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:zff123@127.0.0.1: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:zff123@127.0.0.1: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基本使用

  • 初步使用

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    #! /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:zff123@127.0.0.1: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="tome163@163.com")
    session.add(obj1)

    # 提交事务
    session.commit()
    # 关闭session
    session.close()
  • 原生SQL

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy import create_engine
    from sqlalchemy.sql import text
    from models import *

    engine = create_engine("mysql+pymysql://zff:zff123@127.0.0.1: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': 'zhangsan163@163.com'},
    {'name': 'lisi', 'age': 20, 'value': 'li163@163.com'})
    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()
  • 基本增删改查

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy import create_engine
    from sqlalchemy.sql import text
    from models import *

    engine = create_engine("mysql+pymysql://zff:zff123@127.0.0.1:3306/zff?charset=utf8", max_overflow=0, pool_size=5)
    Session = sessionmaker(bind=engine)

    session = Session()

    # 1. ################ 添加 ################
    """
    obj1 = Users(name="jack", age=19, email="jak163@163.com")
    session.add(obj1)

    session.add_all([
    Users(name="wang", age=19, email="wang163@163.com"),
    Users(name="lucy", age=19, email="lucy@163.com"),
    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()
  • 常用操作

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    # 条件
    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写在哪里,从哪里查。
    使用上面的方法非常麻烦,我们用更高效的方法。
    只要在表里加上这一句话:
    1
    favor = relationship("Favor", backref='pers')
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    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)
  2. 反向查询
    多对一,从一的那端开始查,也就是从没写foreignkey的表里反查。
    多对一,从一的那端反查。
    Person和Favor是多对一,假如查询喜欢蓝色的所有人。Favor的caption为blue的所有对应的Person

传统方式,反向查询:

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

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

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

如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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语句而已。

1
2
3
4
5
6
7
8
9
10
11
12
# 新方式,反向查询
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要成对写在一个表里。

1
2
3
4
5
6
7
8
9
10
11
12
13
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:123@127.0.0.1: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()

1
- 多线程执行示例

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=”shuke”)
session.add(obj1)
session.commit()

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

1
2
- 基于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:123@127.0.0.1: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-武沛齐-博客园](http://www.cnblogs.com/wupeiqi/articles/8259356.html)  
[mysql和SQLAlchemy-博客园](https://www.cnblogs.com/wangqiaomei/p/5721523.html)  
[使用flask-sqlalchemy玩转MySQL | Wing's Tech Space](https://wing324.github.io/2017/02/25/%E4%BD%BF%E7%94%A8flask-sqlalchemy%E7%8E%A9%E8%BD%ACMySQL/)  
[Flask-Migrate的使用 | Wing's Tech Space](https://wing324.github.io/2017/02/26/Flask-Migrate%E7%9A%84%E4%BD%BF%E7%94%A8/)





























本文标题:SqlAlchemy玩转MySQL

文章作者:shuke

发布时间:2020年04月20日 - 15:04

最后更新:2020年04月20日 - 20:04

原始链接:https://shuke163.github.io/2020/04/20/SqlAlchemy%E7%8E%A9%E8%BD%ACMySQL/

许可协议: 署名-非商业性使用-禁止演绎 4.0 国际 转载请保留原文链接及作者。

-------------本文结束感谢您的阅读-------------

本文标题:SqlAlchemy玩转MySQL

文章作者:shuke

发布时间:2020年04月20日 - 15:04

最后更新:2020年04月20日 - 20:04

原始链接:https://shuke163.github.io/2020/04/20/SqlAlchemy%E7%8E%A9%E8%BD%ACMySQL/

许可协议: 署名-非商业性使用-禁止演绎 4.0 国际 转载请保留原文链接及作者。

0%