当前位置：首页 > article >正文

SQLAlchemy系列教程：如何执行原生SQL

article 2025/3/15 1:13:33

Python中的数据库交互提供了高级API。但是，有时您可能需要执行原始SQL以提高效率或利用数据库特定的特性。本指南介绍在SQLAlchemy框架内执行原始SQL。

在SQLAlchemy中执行原生SQL

SQLAlchemy虽然以其对象-关系映射（ORM）功能而闻名，但也允许直接执行原始SQL语句。当您有复杂的查询、需要优化性能或利用数据库引擎特有的特性时，这可能是有益的。执行原始SQL为您提供了这样做的能力和灵活性。

要执行原始SQL，必须使用SQLAlchemy的Connection对象，该对象可以从Engine或Session上下文中获得。让我们通过渐进式示例探索在SQLAlchemy中执行原始SQL的一些常见模式。
在这里插入图片描述

执行简单SQL

要执行原始SQL，需要从引擎获得一个连接：

from sqlalchemy import create_engine
# Replace 'dialect+driver://username:password@host/dbname' with your actual database URI
db_engine = create_engine('dialect+driver://username:password@host/dbname')

with db_engine.connect() as connection:
    result = connection.execute("SELECT * FROM my_table")
    for row in result:
        print(row)

这将打印出‘ my_table ’表结果集中的每一行。

参数化查询

出于安全原因和防止SQL注入攻击，永远不要简单地将变量直接插入到SQL字符串中。相反，使用命名参数或位置占位符：

with db_engine.connect() as connection:
    result = connection.execute(
        "SELECT * FROM users WHERE username = :username", {'username': 'example_user'}
    )
    user = result.fetchone()
    print(user)

在上面的示例中，“:username ”是一个占位符，可以被“ example_user ”安全地替换。

使用文本SQL

SQLAlchemy的text函数可以用来创建带有占位符的SQL表达式：

from sqlalchemy.sql import text

sql = text("SELECT * FROM users WHERE username = :username")

with db_engine.connect() as connection:
    result = connection.execute(sql, username='example_user')
    user = result.fetchone()
    print(user)

这里，文本函数用命名参数包装SQL，提供灵活性和注入预防。

执行插入、更新、删除

INSERT、UPDATE、DELETE等修改操作也可以用类似的方式执行：

# Inserting a new user
insert_sql = text("INSERT INTO users (username, email) VALUES (:username, :email)")

with db_engine.connect() as connection:
    connection.execute(insert_sql, username='new_user', email='new_user@example.com')

# Updating a user's email
update_sql = text("UPDATE users SET email = :email WHERE username = :username")

with db_engine.connect() as connection:
    connection.execute(update_sql, email='updated_user@example.com', username='existing_user')

# Deleting a user
delete_sql = text("DELETE FROM users WHERE username = :username")

with db_engine.connect() as connection:
    connection.execute(delete_sql, username='obsolete_user')

处理事务

事务处理使用连接对象，在执行SQL语句之前首先开始一个事务。这确保了原子性：

with db_engine.connect() as connection:
    transaction = connection.begin()
    try:
        connection.execute(insert_sql, {...})
        connection.execute(update_sql, {...})
        transaction.commit()
    except:
        transaction.rollback()
        raise

这将插入和更新操作包装在一个事务中，该事务可以在失败时回滚。

执行存储过程

存储过程也可以通过原始SQL调用：

call_procedure_sql = text("CALL my_stored_procedure(:param)")

with db_engine.connect() as connection:
    result = connection.execute(call_procedure_sql, param='value')
    for row in result:
        print(row)

使用SQLAlchemy Core进行复杂查询

除了简单的文本语句，SQLAlchemy的核心语言还可以将文本SQL与Python逻辑相结合：

from sqlalchemy.sql import select, table, column

t_user = table('users', column('username'), column('email'))
stmt = select([t_user]).where(t_user.c.username == 'example_user')

with db_engine.connect() as connection:
    for row in connection.execute(stmt):
        print(row)

这个示例演示了如何使用SQLAlchemy Core构造从用户名匹配“example_user”的“users”表中进行选择。

访问本地数据库功能

最后，使用SQLAlchemy，在需要特定于数据库功能的情况下，您可以将原始SQL直接传递给底层DBAPI连接：

with db_engine.raw_connection() as raw_conn:
    cursor = raw_conn.cursor()
    cursor.execute("YOUR_VENDOR_SPECIFIC_SQL_HERE")
    results = cursor.fetchall()
    for result in results:
        print(result)
    cursor.close()