Python 程式連線
Python 通常透過 PostgreSQL 驅動程式連線資料庫，常用套件是 **`psycopg`**（新版，原名 psycopg3）。
## 1. 安裝 PostgreSQL 驅動程式
```bash
pip install "psycopg[binary]"
```
也可以使用較舊的 `psycopg2`：
```bash
pip install psycopg2-binary
```
以下以新版 `psycopg` 為例。
---
## 2. 建立資料庫連線
```python
import psycopg
conn = psycopg.connect(
host="localhost",
port=5432,
dbname="mydb",
user="postgres",
password="mypassword"
)
print("資料庫連線成功")
conn.close()
```
常見連線參數：
| 參數 | 說明 |
|---|---|
| `host` | PostgreSQL 伺服器位址 |
| `port` | 預設為 `5432` |
| `dbname` | 資料庫名稱 |
| `user` | 使用者名稱 |
| `password` | 密碼 |
也可以使用連線字串：
```python
conn = psycopg.connect(
"postgresql://postgres:mypassword@localhost:5432/mydb"
)
```
正式環境不建議把密碼直接寫在程式碼中，應使用環境變數：
```python
import os
import psycopg
conn = psycopg.connect(
host=os.getenv("PGHOST", "localhost"),
port=os.getenv("PGPORT", "5432"),
dbname=os.getenv("PGDATABASE", "mydb"),
user=os.getenv("PGUSER", "postgres"),
password=os.getenv("PGPASSWORD")
)
```
---
## 3. 建立資料表並執行 SQL
透過 `cursor()` 建立游標，再使用 `execute()` 執行 SQL：
```python
import psycopg
with psycopg.connect(
host="localhost",
port=5432,
dbname="mydb",
user="postgres",
password="mypassword"
) as conn:
with conn.cursor() as cur:
cur.execute("""
CREATE TABLE IF NOT EXISTS users (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
email VARCHAR(200) UNIQUE NOT NULL
)
""")
# with 區塊結束時，若沒有例外會自動 commit
```
使用 `with` 可以自動處理：
- 關閉游標
- 提交交易 `commit`
- 發生錯誤時回滾 `rollback`
- 關閉資料庫連線
---
## 4. 新增資料
### 使用參數化查詢
```python
import psycopg
with psycopg.connect(
host="localhost",
dbname="mydb",
user="postgres",
password="mypassword"
) as conn:
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO users (name, email)
VALUES (%s, %s)
""",
("王小明", "ming@example.com")
)
```
`%s` 是參數佔位符，實際值放在第二個參數中。
不要使用字串串接 SQL：
```python
# 不建議，可能造成 SQL Injection
name = "王小明"
sql = f"INSERT INTO users (name) VALUES ('{name}')"
```
應使用參數化方式：
```python
cur.execute(
"INSERT INTO users (name) VALUES (%s)",
(name,)
)
```
注意：單一參數的 tuple 必須加逗號，例如：
```python
(name,)
```
---
## 5. 查詢資料
使用 `SELECT` 查詢資料後，可以使用不同方法取得結果。
### 取得一筆資料：`fetchone()`
```python
with psycopg.connect(
host="localhost",
dbname="mydb",
user="postgres",
password="mypassword"
) as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT id, name, email FROM users WHERE id = %s",
(1,)
)
row = cur.fetchone()
if row is not None:
print("ID:", row[0])
print("姓名:", row[1])
print("Email:", row[2])
else:
print("找不到資料")
```
查詢結果通常是 tuple：
```python
(1, '王小明', 'ming@example.com')
```
---
### 取得多筆資料：`fetchall()`
```python
with psycopg.connect(
host="localhost",
dbname="mydb",
user="postgres",
password="mypassword"
) as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT id, name, email FROM users ORDER BY id"
)
rows = cur.fetchall()
for row in rows:
print(row)
```
---
### 逐筆讀取資料
資料量很大時，不一定要一次使用 `fetchall()`，可以直接迭代 cursor：
```python
with psycopg.connect(
host="localhost",
dbname="mydb",
user="postgres",
password="mypassword"
) as conn:
with conn.cursor() as cur:
cur.execute("SELECT id, name, email FROM users")
for row in cur:
user_id, name, email = row
print(user_id, name, email)
```
---
### 取得欄位名稱
如果希望取得字典格式的資料，可以使用 `dict_row`：
```python
import psycopg
from psycopg.rows import dict_row
with psycopg.connect(
host="localhost",
dbname="mydb",
user="postgres",
password="mypassword",
row_factory=dict_row
) as conn:
with conn.cursor() as cur:
cur.execute("SELECT id, name, email FROM users")
rows = cur.fetchall()
for row in rows:
print(row["id"], row["name"], row["email"])
```
結果類似：
```python
{
"id": 1,
"name": "王小明",
"email": "ming@example.com"
}
```
---
## 6. 更新與刪除資料
### 更新資料
```python
with psycopg.connect(
host="localhost",
dbname="mydb",
user="postgres",
password="mypassword"
) as conn:
with conn.cursor() as cur:
cur.execute(
"""
UPDATE users
SET email = %s
WHERE id = %s
""",
("new@example.com", 1)
)
print("更新筆數:", cur.rowcount)
```
### 刪除資料
```python
with psycopg.connect(
host="localhost",
dbname="mydb",
user="postgres",
password="mypassword"
) as conn:
with conn.cursor() as cur:
cur.execute(
"DELETE FROM users WHERE id = %s",
(1,)
)
print("刪除筆數:", cur.rowcount)
```
---
## 7. 手動處理交易
如果不使用 `with`，需要自行提交或回滾：
```python
import psycopg
conn = psycopg.connect(
host="localhost",
dbname="mydb",
user="postgres",
password="mypassword"
)
try:
with conn.cursor() as cur:
cur.execute(
"INSERT INTO users (name, email) VALUES (%s, %s)",
("李小華", "hua@example.com")
)
cur.execute(
"UPDATE users SET name = %s WHERE email = %s",
("李小華", "hua@example.com")
)
conn.commit()
print("交易成功")
except Exception as e:
conn.rollback()
print("交易失敗，已回滾:", e)
finally:
conn.close()
```
交易原則：
- `commit()`：確認並保存修改
- `rollback()`：取消尚未提交的修改
- `close()`：關閉連線
---
## 8. 批次新增資料
可以使用 `executemany()`：
```python
users = [
("小美", "mei@example.com"),
("小強", "qiang@example.com"),
("小芳", "fang@example.com"),
]
with psycopg.connect(
host="localhost",
dbname="mydb",
user="postgres",
password="mypassword"
) as conn:
with conn.cursor() as cur:
cur.executemany(
"""
INSERT INTO users (name, email)
VALUES (%s, %s)
""",
users
)
```
---
## 9. 完整範例
```python
import psycopg
from psycopg.rows import dict_row
DB_CONFIG = {
"host": "localhost",
"port": 5432,
"dbname": "mydb",
"user": "postgres",
"password": "mypassword",
"row_factory": dict_row
}
def main():
with psycopg.connect(**DB_CONFIG) as conn:
with conn.cursor() as cur:
cur.execute("""
CREATE TABLE IF NOT EXISTS products (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
price NUMERIC(10, 2) NOT NULL
)
""")
cur.execute(
"""
INSERT INTO products (name, price)
VALUES (%s, %s)
RETURNING id
""",
("鍵盤", 1290.00)
)
product_id = cur.fetchone()["id"]
print("新增產品 ID:", product_id)
cur.execute(
"""
SELECT id, name, price
FROM products
WHERE price >= %s
ORDER BY id
""",
(1000,)
)
products = cur.fetchall()
for product in products:
print(product)
if __name__ == "__main__":
main()
```
這個範例示範了：
1. 連線 PostgreSQL
2. 建立資料表
3. 新增資料
4. 使用 `RETURNING` 取得新增資料的 ID
5. 執行查詢
6. 使用 `fetchall()` 取得結果
7. 以字典格式讀取資料
8. 自動提交交易並關閉連線
重點是：使用 `cursor.execute()` 執行 SQL、使用 `fetchone()` 或 `fetchall()` 取得查詢結果，並一律優先使用參數化查詢以避免 SQL Injection。
相關學習地圖、教學課程
Python 資料工程