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

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

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

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

執(zhí)行簡(jiǎn)單SQL

要執(zhí)行原始SQL，需要從引擎獲得一個(gè)連接：

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)不要簡(jiǎ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 ”是一個(gè)占位符，可以被“ example_user ”安全地替換。

使用文本SQL

SQLAlchemy的text函數(shù)可以用來(lái)創(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等修改操作也可以用類(lèi)似的方式執(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ù)處理使用連接對(duì)象，在執(zhí)行SQL語(yǔ)句之前首先開(kāi)始一個(gè)事務(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

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

執(zhí)行存儲(chǔ)過(guò)程

存儲(chǔ)過(guò)程也可以通過(guò)原始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ù)雜查詢

除了簡(jiǎn)單的文本語(yǔ)句，SQLAlchemy的核心語(yǔ)言還可以將文本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)

這個(gè)示例演示了如何使用SQLAlchemy Core構(gòu)造從用戶名匹配“example_user”的“users”表中進(jìn)行選擇。

訪問(wèn)本地?cái)?shù)據(jù)庫(kù)功能

最后，使用SQLAlchemy，在需要特定于數(shù)據(jù)庫(kù)功能的情況下，您可以將原始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的各種方法，從簡(jiǎn)單查詢到復(fù)雜事務(wù)，甚至直接訪問(wèn)DB API功能。負(fù)責(zé)任地使用這些方法，始終將查詢參數(shù)化以防止SQL注入，并記住盡可能利用SQLAlchemy健壯的ORM特性。