SqlAlchemy玩轉MySQL

本文轉載自查看原文 2020-09-02 14:15 542 Python

一. ORM框架

Web 開發中，一個重要的組成部分便是數據庫了。Web 程序中最常用的莫過於關系型數據庫了，也稱 SQL 數據庫。另外，文檔數據庫（如 mongodb）、鍵值對數據庫（如 redis）近幾年也逐漸在 web 開發中流行起來，我們習慣把這兩種數據庫稱為 NoSQL 數據庫。
大多數的關系型數據庫引擎（比如 MySQL、Postgres 和 SQLite）都有對應的 Python 包。在這里，我們不直接使用這些數據庫引擎提供的 Python 包，而是使用對象關系映射（Object-Relational Mapper, ORM）框架，它將低層的數據庫操作指令抽象成高層的面向對象操作。也就是說，如果我們直接使用數據庫引擎，我們就要寫 SQL 操作語句，但是，如果我們使用了 ORM 框架，我們對諸如表、文檔此類的數據庫實體就可以簡化成對 Python 對象的操作。
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

三. 框架基本結構

組成部分：
Engine，框架的引擎
Connection Pooling ，數據庫連接池
Dialect，選擇連接數據庫的DB API種類
Schema/Types，架構和類型
SQL Exprression Language，SQL表達式語言

四. 數據庫連接串

數據庫引擎	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...]

SQLAlchemy本身無法操作數據庫，其必須以來pymsql等第三方插件，Dialect用於和數據API進行交流，根據配置文件的不同調用不同的數據庫API，從而實現對數據庫的操作，下面的表格中，username 和 password 表示登錄數據庫的用戶名和密碼，hostname 表示 SQL 服務所在的主機，可以是本地主機（localhost）也可以是遠程服務器，database 表示要使用的數據庫。有一點需要注意的是，SQLite 數據庫不需要使用服務器，它使用硬盤上的文件名作為 database。如：

數據庫引擎	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...]

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

方式二

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

方式三

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 創建數據庫表

單表示例

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示例

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基本使用

初步使用

#! /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

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

基本增刪改查

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

常用操作

# 條件
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)

正向查詢
多對一，多的那端開始查，也就是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)

反向查詢
多對一，從一的那端開始查，也就是從沒寫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: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()

多線程執行示例

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

基於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-武沛齊-博客園
 mysql和SQLAlchemy-博客園
 使用flask-sqlalchemy玩轉MySQL | Wing's Tech Space
Flask-Migrate的使用 | Wing's Tech Space

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 mysql八：ORM框架SQLAlchemy Python操作MySQL之SQLAlchemy 【python】-- SQLAlchemy操作MySQL python sqlalchemy mysql 自動映射【MySQL】玩轉定時器 sqlalchemy 使用pymysql連接mysql 1366錯誤利用sqlalchemy(ORM框架)連接操作mysql_mysql 使用mysql+sqlalchemy中文亂碼問題（mysql5.7+） Flask 操作Mysql數據庫 - flask-sqlalchemy擴展 python3 + flask + sqlalchemy +orm(1):鏈接mysql 數據庫