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

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

article 2026/3/17 8:35:13

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 textsql = 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, columnt_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()

最后总结

本指南重点介绍了使用SQLAlchemy执行原始SQL的各种方法，从简单查询到复杂事务，甚至直接访问DB API功能。负责任地使用这些方法，始终将查询参数化以防止SQL注入，并记住尽可能利用SQLAlchemy健壮的ORM特性。