Python中的數(shù)據(jù)庫交互提供了高級API。但是，有時您可能需要執(zhí)行原始SQL以提高效率或利用數(shù)據(jù)庫特定的特性。本指南介紹在SQLAlchemy框架內(nèi)執(zhí)行原始SQL。

在SQLAlchemy中執(zhí)行原生SQL

SQLAlchemy雖然以其對象-關(guān)系映射（ORM）功能而聞名，但也允許直接執(zhí)行原始SQL語句。當(dāng)您有復(fù)雜的查詢、需要優(yōu)化性能或利用數(shù)據(jù)庫引擎特有的特性時，這可能是有益的。執(zhí)行原始SQL為您提供了這樣做的能力和靈活性。

要執(zhí)行原始SQL，必須使用SQLAlchemy的Connection對象，該對象可以從Engine或Session上下文中獲得。讓我們通過漸進(jìn)式示例探索在SQLAlchemy中執(zhí)行原始SQL的一些常見模式。

執(zhí)行簡單SQL

要執(zhí)行原始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 ’表結(jié)果集中的每一行。

參數(shù)化查詢

出于安全原因和防止SQL注入攻擊，永遠(yuǎn)不要簡單地將變量直接插入到SQL字符串中。相反，使用命名參數(shù)或位置占位符：

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函數(shù)可以用來創(chuàng)建帶有占位符的SQL表達(dá)式：

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)

這里，文本函數(shù)用命名參數(shù)包裝SQL，提供靈活性和注入預(yù)防。

執(zhí)行插入、更新、刪除

INSERT、UPDATE、DELETE等修改操作也可以用類似的方式執(zhí)行：

# 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')

處理事務(wù)

事務(wù)處理使用連接對象，在執(zhí)行SQL語句之前首先開始一個事務(wù)。這確保了原子性：

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

這將插入和更新操作包裝在一個事務(wù)中，該事務(wù)可以在失敗時回滾。

執(zhí)行存儲過程

存儲過程也可以通過原始SQL調(diào)用：

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進(jìn)行復(fù)雜查詢

除了簡單的文本語句，SQLAlchemy的核心語言還可以將文本SQL與Python邏輯相結(jié)合：

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構(gòu)造從用戶名匹配“example_user”的“users”表中進(jìn)行選擇。

訪問本地數(shù)據(jù)庫功能

最后，使用SQLAlchemy，在需要特定于數(shù)據(jù)庫功能的情況下，您可以將原始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()

最后總結(jié)

本指南重點(diǎn)介紹了使用SQLAlchemy執(zhí)行原始SQL的各種方法，從簡單查詢到復(fù)雜事務(wù)，甚至直接訪問DB API功能。負(fù)責(zé)任地使用這些方法，始終將查詢參數(shù)化以防止SQL注入，并記住盡可能利用SQLAlchemy健壯的ORM特性。