SQLAlchemy 执行原生 SQL语句

使用 sqlalchemy ,一般通过 Session 对象 ORM 方式操作数据库。如果需要通过 原生 SQL 语句操作数据库,就需要跟 Engine 和 Connect 对象打交道。

Engine 对象包含数据库连接池和数据库方言,通过 create_engine() 函数来创建,engine 对象的 connect() 方法返回 Connection 对象,Connection 对象提供 execute() 方法,允许通过原生 sql 语句来操作数据库。

本篇以 SQLite 数据库为例,简单介绍原生 SQL 操作的一些要点。

原生 SQL 语句

  • create_engine() 函数创建 Engine 对象
  • Engine.connect() 方法返回 Connection 对象
  • Connection.execute() 方法执行 SQL 语句
  • Connection.close() 方法释放资源

示例:

# -*- encoding: utf-8 -*-
from sqlalchemy import create_engine

engine = create_engine('sqlite:///testdb.db')
def test_select_statement():
    with engine.connect() as conn:
        result_proxy = conn.execute("select * from employees") # 返回值为ResultProxy类型
        result = result_proxy.fetchall()
        for item in result: print(item)

带有参数的 SQL 语句

SQLAlchemy 支持两种格式的 sql 语句:?和 :number。

# -*- encoding: utf-8 -*-
from sqlalchemy import create_engine

def test_parameter_method1():
    with engine.connect() as conn:
        conn.execute(
        """INSERT INTO employees
            (EMP_ID, FIRST_NAME, LAST_NAME, GENDER,AGE, EMAIL, PHONE_NR,EDUCATION,MARITAL_STAT, NR_OF_CHILDREN)
            VALUES (?,?,?,?,?,?,?,?,?,?);
        """,
            ('9002', 'Stone2', 'Wang', 'M', 20,
        'stone@gmail.com', '138xxx', 'Bachelor', 'Single', 0)
        )

方法 2 :

# -*- encoding: utf-8 -*-
from sqlalchemy import create_engine

def test_parameter_method2():
    with engine.connect() as conn:
        conn.execute(
            """INSERT INTO employees
                (EMP_ID, FIRST_NAME, LAST_NAME, GENDER,
                AGE, EMAIL, PHONE_NR, EDUCATION,
                MARITAL_STAT, NR_OF_CHILDREN)
                VALUES (:1,:2,:3,:4,:5,:6,:7,:8,:9,:10);
            """,
            ('9003', 'Stone3', 'Wang', 'M', 20, 'stone@gmail.com',
            '138xxx', 'Bachelor', 'Single', 0)
        )

插入多行

SQLAlchemy 支持一次插入多行,需要插入的数据放在 list 中:

# -*- encoding: utf-8 -*-
from sqlalchemy import create_engine

def test_insert_multiple_rows(self):
    with engine.connect() as conn:
        values = [
            ('9004', 'Stone4', 'Wang', 'M', 20, 'stone@gmail.com', '138xxx', 'Bachelor', 'Single', 0),
            ('9005', 'Stone5', 'Wang', 'M', 20, 'stone@gmail.com', '138xxx', 'Bachelor', 'Single', 0),
            ('9006', 'Stone6', 'Wang', 'M', 20, 'stone@gmail.com', '138xxx', 'Bachelor', 'Single', 0)
        ]
    conn.execute(
        """INSERT INTO employees
        (EMP_ID, FIRST_NAME, LAST_NAME, GENDER,AGE,
         EMAIL, PHONE_NR, EDUCATION,MARITAL_STAT, NR_OF_CHILDREN)
         VALUES (?,?,?,?,?,?,?,?,?,?);
        """, values)

事务操作

由于执行 sql 插入操作自动提交 (commit),sqlalchemy 提供了 Transactions 来管理 commit 和 rollback,需要提交的时候用 commit() 方法,需要回滚的时候用 rollback() 方法。

def test_txn(self):
conn = engine.connect()
with conn.begin() as txn:
conn.execute(
"""INSERT INTO employees
(EMP_ID, FIRST_NAME, LAST_NAME, GENDER, AGE, EMAIL, PHONE_NR,
EDUCATION, MARITAL_STAT, NR_OF_CHILDREN)
VALUES (:1,:2,:3,:4,:5,:6,:7,:8,:9,:10);
""",
('9007', 'Stone7', 'Wang', 'M', 20, 'stone@gmail.com', '138xxx', 'Bachelor', 'Single', 0)
)
txn.commit() conn.close()

源码

github – executing raw sql statement