Python 程式設計/資料庫
Python 透過簡單的 API 支援與資料庫互動。Python 自帶的模組包括用於 SQLite 和 Berkeley DB 的模組。用於 MySQL、PostgreSQL、FirebirdSQL 等其他資料庫的模組作為第三方模組提供。後者需要在使用前下載並安裝。例如，可以使用 Debian 包“python-mysqldb”安裝 MySQLdb 包。
使用 MySQL 的示例如下所示
import MySQLdb
db = MySQLdb.connect("host machine", "dbuser", "password", "dbname")
cursor = db.cursor()
query = """SELECT * FROM sampletable"""
lines = cursor.execute(query)
data = cursor.fetchall()
db.close()
在第一行，模組MySQLdb被匯入。然後建立與資料庫的連線，在第 4 行，我們將要執行的實際 SQL 語句儲存在變數query中。在第 5 行，我們執行查詢，在第 6 行，我們獲取所有資料。在這段程式碼執行之後，lines包含獲取的行數（例如，表中的行數sampletable）。變數data包含所有實際資料，例如內容sampletable。最後，連線到資料庫將再次關閉。如果行數很大，最好使用row = cursor.fetchone()並分別處理這些行
#first 5 lines are the same as above
while True:
row = cursor.fetchone()
if row == None: break
#do something with this row of data
db.close()
顯然，必須對行使用某種資料處理，否則資料將不會被儲存。的輸出結果fetchone()命令是 元組。
為了使連線的初始化更容易，可以使用配置檔案
import MySQLdb
db = MySQLdb.connect(read_default_file="~/.my.cnf")
...
在這裡，主目錄中的 .my.cnf 檔案包含 MySQL 所需的配置資訊。
SQLite 的示例與上面的示例非常相似，並且遊標提供了許多相同的功能。
import sqlite3
db = sqlite3.connect("/path/to/file")
cursor = db.cursor()
query = """SELECT * FROM sampletable"""
lines = cursor.execute(query)
data = cursor.fetchall()
db.close()
寫入資料庫時，必須記住呼叫 db.commit()，否則更改不會儲存
import sqlite3
db = sqlite3.connect("/path/to/file")
cursor = db.cursor()
query = """INSERT INTO sampletable (value1, value2) VALUES (1,'test')"""
cursor.execute(query)
db.commit()
db.close()
import psycopg2
conn = psycopg2.connect("dbname=test")
cursor = conn.cursor()
cursor.execute("select * from test");
for i in cursor.next():
print(i)
conn.close()
import firebirdsql
conn = firebirdsql.connect(dsn='localhost/3050:/var/lib/firebird/2.5/test.fdb', user='alice', password='wonderland')
cur = conn.cursor()
cur.execute("select * from baz")
for c in cur.fetchall():
print(c)
conn.close()
您經常需要將動態資料替換到查詢字串中。務必確保正確執行此操作。
# Do not do this!
result = db.execute("SELECT name FROM employees WHERE location = '" + location + "'")
此示例錯誤，因為它沒有正確處理要替換的字串中的特殊字元（如撇號）。如果您的程式碼必須處理潛在的惡意使用者（例如在面向公眾的 Web 伺服器上），這可能會讓您面臨SQL 注入攻擊的風險。
對於簡單情況，請使用 execute 方法提供的自動引數替換，例如
result = db.execute("SELECT name FROM employees WHERE location = ?", [location])
DBMS 介面本身會自動將您傳遞的值轉換為正確的 SQL 語法。
對於更復雜的情況，DBMS 模組應提供一個您可以顯式呼叫的引用函式。例如，MySQLdb 提供 escape_string 方法，而 APSW（用於 SQLite3）提供 format_sql_value。這在查詢結構採用更動態形式時是必要的
criteria = [("company", company)] # list of tuples (fieldname, value)
if department != None :
criteria.append(("department", department))
# ... append other optional criteria as appropriate ...
result = db.execute(
"SELECT name FROM employees WHERE "
+
" and ".join(
"%s = %s" % (criterion[0], MySQLdb.escape_string(criterion[1]))
for criterion in criteria
)
)
這將動態構建查詢，例如“select name from employees where company = 'some company'”或“select name from employees where company = 'some company' and department = 'some department'”，具體取決於使用者填寫了哪些欄位。
Python 迭代器非常適合迭代大量資料庫記錄的問題。以下是一個函式示例，該函式執行資料庫查詢並返回結果的迭代器，而不是一次返回所有結果。它依賴於以下事實：在 APSW（SQLite 的 Python 3 介面庫）中，cursor.execute 方法本身返回結果記錄的迭代器。結果是您可以編寫非常簡潔的程式碼來執行 Python 中複雜的資料庫查詢。
def db_iter(db, cmd, mapfn = lambda x : x) :
"executes cmd on a new cursor from connection db and yields the results in turn."
cu = db.cursor()
result = cu.execute(cmd)
while True:
yield mapfn(next(result))
此函式的示例用法
for artist, publisher in db_iter(
db = db,
cmd =
"SELECT artist, publisher FROM artists WHERE location = %s"
%
apsw.format_sql_value(location)
):
print(artist, publisher)
和
for location in db_iter(
db = db,
cmd = "SELECT DISTINCT location FROM artists",
mapfn = lambda x : x[0]
):
print(location)
在第一個示例中，由於 db_iter 為每個記錄返回一個元組，因此可以直接將其分配給記錄欄位的各個變數。在第二個示例中，元組只有一個元素，因此使用自定義的 mapfn 來提取此元素並返回它，而不是返回元組。
資料庫表定義經常會發生變化。隨著應用程式需求的發展，欄位甚至整個表經常會被新增，或者有時會被刪除。考慮以下語句
result = db.execute("select * from employees")
您可能碰巧知道employees表當前包含，比如說，4 個欄位。但是明天有人可能會新增第五個欄位。您是否記得更新程式碼以處理這種情況？如果沒有，它可能會崩潰。或者更糟糕的是，產生錯誤的結果！
最好始終列出您感興趣的特定欄位，無論有多少個
result = db.execute("select name, address, department, location from employees")
這樣，新增的任何額外欄位都將被簡單地忽略。如果刪除了任何命名的欄位，程式碼至少會因執行時錯誤而失敗，這提醒您忘記更新它！
考慮以下場景：您的銷售公司資料庫有一個員工表，還有一個每個員工進行的銷售記錄表。您希望遍歷這些銷售條目，並生成一些每個員工的統計資訊。一種天真的方法可能是
- 查詢資料庫以獲取員工列表
- 對於每個員工，執行資料庫查詢以獲取每個員工的銷售列表。
如果您有很多員工，那麼第一個查詢可能會產生一個很大的列表，並且第二步將涉及相應數量的資料庫查詢。
事實上，整個處理迴圈可以透過單個數據庫查詢執行，使用稱為 join 的標準 SQL 結構。
以下是此類迴圈的示例
rows = db_iter \
(
db = db,
cmd =
"select employees.name, sales.amount, sales.date from"
" employees left join sales on employees.id = sales.employee_id"
" order by employees.name, sales.date"
)
prev_employee_name = None
while True:
row = next(rows, None)
if row != None :
employee_name, amount, date = row
if row == None or employee_name != prev_employee_name :
if prev_employee_name != None :
# done stats for this employee
report(prev_employee_name, employee_stats)
if row == None :
break
# start stats for a new employee
prev_employee_name = employee_name
employee_stats = {"total_sales" : 0, "number_of_sales" : 0}
if date != None :
employee_stats["earliest_sale"] = date
# another row of stats for this employee
if amount != None :
employee_stats["total_sales"] += amount
employee_stats["number_of_sales"] += 1
if date != None :
employee_stats["latest_sale"] = date
這裡的統計資訊非常簡單：最早和最晚的銷售、銷售數量和總金額，並且可以直接在 SQL 查詢中計算。但是，相同的迴圈可以計算更復雜的統計資訊（如標準差），這些統計資訊不能直接在簡單的 SQL 查詢中表示。
請注意，每個員工的統計資訊是如何在以下兩種情況之一下輸出的
- 下一條記錄的員工姓名與上一條記錄不同
- 查詢結果已結束。
這兩個條件使用`row == None or employee_name != prev_employee_name`進行測試；在輸出員工統計資訊後，使用單獨的第二個條件檢查`row == None`來終止迴圈。如果迴圈沒有終止，則為新員工初始化處理過程。
還要注意在此處使用了`left join`：如果某個員工沒有銷售記錄，則聯接將返回該員工的單行記錄，其中來自銷售表的欄位將為SQL `null`值（在Python中表示為`None`）。這就是為什麼在處理這些欄位之前需要檢查此類`None`值的原因。
或者，我們可以使用`inner join`，它對於沒有銷售記錄的員工將不返回任何結果。是否要從報表中省略此類員工，或者將他們包含在內並顯示為零總計，這完全取決於您的應用程式。
- APSW 模組，code.google.com — Python 2.x 和 3.x 的 SQLite3
- SQLite 文件
- Psycopg2（PostgreSQL 模組 - 更新），initd.org
- PyGreSQL（PostgreSQL 模組 - 舊版），pygresql.org
- MySQL 模組，sourceforge.net
- FirebirdSQL 模組，github.com