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特性。