從網址取得YouBike與開放資料——requests安全入門
學會向公開資料服務提出請求、看懂回應,並在網路、格式或欄位改變時安全停止。
完成本篇後,你能使用
requests.get() 取得JSON,理解HTTP狀態碼、標頭與逾時,使用 raise_for_status() 和例外處理,驗證資料結構,建立節制的重試策略,並保存來源與擷取時間。一、這次使用哪一份資料?
本篇採用臺北市政府交通局提供的「YouBike2.0臺北市公共自行車即時資訊」作為教學案例。政府資料開放平臺列出的資料格式為JSON,欄位包括站點名稱、行政區、更新時間、可借車輛數、可還空位數及經緯度等,授權方式為政府資料開放授權條款第1版。
二、requests不是標準函式庫
requests 是第三方套件。在Thonny或本機環境可安裝;Google Colab通常已具備,但仍應先確認。
python -m pip install requestsPython也提供標準函式庫 urllib.request;本篇先使用介面較直觀的requests,之後再比較兩者。
三、第一個GET請求
import requests
URL = (
"https://tcgbusfs.blob.core.windows.net/"
"dotapp/youbike/v2/youbike_immediate.json"
)
response = requests.get(URL, timeout=(3.05, 15))
response.raise_for_status()
data = response.json()
print(type(data))
print("資料筆數:", len(data))timeout=(3.05, 15) 分別限制建立連線與等待資料的時間。依requests官方文件,若未明確設定timeout,請求預設不會自動逾時;timeout也不是整份下載的總秒數,而是網路操作等待限制。
四、回應物件裡有什麼?
print("狀態碼:", response.status_code)
print("內容類型:", response.headers.get("Content-Type"))
print("回應位元組:", len(response.content))
print("最終網址:", response.url)| 狀態碼範圍 | 一般意義 | 程式行動 |
|---|---|---|
| 2xx | 請求成功 | 仍要驗證格式與欄位 |
| 3xx | 重新導向 | 確認最終網址是否符合預期 |
| 4xx | 請求、權限或資源問題 | 不要盲目重試,先檢查設定 |
| 5xx | 伺服器暫時或內部問題 | 可有限度重試並保留紀錄 |
五、完整處理網路與JSON錯誤
import requests
def fetch_json(url):
try:
response = requests.get(
url,
timeout=(3.05, 15),
headers={"User-Agent": "NFU-Python-USR-Education/1.0"},
)
response.raise_for_status()
return response.json()
except requests.Timeout:
print("資料服務逾時,請稍後再試。")
except requests.HTTPError as error:
print("HTTP狀態錯誤:", error)
except requests.ConnectionError:
print("無法連線,請檢查網路或資料來源。")
except requests.JSONDecodeError:
print("回應不是有效JSON,請確認服務格式。")
except requests.RequestException as error:
print("其他請求錯誤:", error)
return None不要用空白 except: 隱藏問題。錯誤發生時也不要把上一次資料假裝成最新資料;若使用快取,畫面須清楚標示擷取時間與資料可能過期。
六、先確認最外層資料型別
def validate_dataset(data):
if not isinstance(data, list):
raise TypeError("預期JSON最外層是串列")
if not data:
raise ValueError("資料集是空串列")
if not all(isinstance(item, dict) for item in data):
raise TypeError("每一筆站點資料必須是字典")
return dataresponse.json() 成功只表示回應可解析為JSON,不代表內容一定符合教學程式預期。
七、欄位名稱也要驗證
REQUIRED_FIELDS = {
"sno",
"sna",
"sarea",
"available_rent_bikes",
"available_return_bikes",
"updateTime",
}
def parse_station(item):
missing = REQUIRED_FIELDS - item.keys()
if missing:
names = "、".join(sorted(missing))
raise KeyError(f"缺少欄位:{names}")
return {
"站點編號": str(item["sno"]),
"站點名稱": str(item["sna"]).strip(),
"行政區": str(item["sarea"]).strip(),
"可借車輛": int(item["available_rent_bikes"]),
"可還空位": int(item["available_return_bikes"]),
"更新時間": str(item["updateTime"]),
}八、單筆錯誤不拖垮整批資料
def parse_stations(data):
valid = []
errors = []
for index, item in enumerate(data, start=1):
try:
station = parse_station(item)
if station["可借車輛"] < 0:
raise ValueError("可借車輛不可為負數")
if station["可還空位"] < 0:
raise ValueError("可還空位不可為負數")
valid.append(station)
except (KeyError, TypeError, ValueError) as error:
errors.append({"資料序號": index, "原因": str(error)})
return valid, errors九、篩選可借車站點
def find_available_stations(stations, area=None, minimum=1):
result = []
for station in stations:
if area is not None and station["行政區"] != area:
continue
if station["可借車輛"] >= minimum:
result.append(station)
return sorted(
result,
key=lambda item: item["可借車輛"],
reverse=True,
)這只反映資料擷取當下的回應,不保證使用者抵達時仍有車。導覽或交通建議應顯示更新時間,避免把即時資料誤寫成長期事實。
十、把來源與時間一起保存
import json
from datetime import datetime, timezone
from pathlib import Path
def save_snapshot(path, source_url, stations, errors):
snapshot = {
"source_url": source_url,
"fetched_at_utc": datetime.now(
timezone.utc
).isoformat(),
"valid_count": len(stations),
"error_count": len(errors),
"stations": stations,
"errors": errors,
}
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("x", encoding="utf-8") as file:
json.dump(snapshot, file, ensure_ascii=False, indent=2)
return path使用 x 避免覆蓋快照。檔案名稱可加入時間戳;日後分析時才知道資料是哪一刻取得、有效與錯誤各多少筆。
十一、有限度重試GET請求
GET通常是冪等操作,可對連線錯誤、429或部分5xx狀態有限度重試;不要無限迴圈,也不要對所有4xx重試。
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def build_session():
retry = Retry(
total=3,
connect=3,
read=2,
backoff_factor=0.5,
status_forcelist=(429, 500, 502, 503, 504),
allowed_methods=frozenset({"GET"}),
respect_retry_after_header=True,
)
adapter = HTTPAdapter(max_retries=retry)
session = requests.Session()
session.mount("https://", adapter)
return sessionurllib3官方文件說明,是否針對特定狀態碼重試由 status_forcelist 與允許方法共同決定。若服務提供 Retry-After,應尊重伺服器要求。
十二、限制下載大小
來源即使可信也可能異常回傳超大內容。串流下載時可設定上限:
import requests
def download_limited(url, max_bytes=10_000_000):
chunks = []
total = 0
with requests.get(
url, timeout=(3.05, 15), stream=True
) as response:
response.raise_for_status()
for chunk in response.iter_content(chunk_size=64 * 1024):
if not chunk:
continue
total += len(chunk)
if total > max_bytes:
raise ValueError("回應超過允許大小")
chunks.append(chunk)
return b"".join(chunks)十三、整合主程式
def main():
data = fetch_json(URL)
if data is None:
return
try:
validate_dataset(data)
except (TypeError, ValueError) as error:
print("資料集結構錯誤:", error)
return
stations, errors = parse_stations(data)
print("有效站點:", len(stations))
print("錯誤資料:", len(errors))
available = find_available_stations(
stations, minimum=5
)
for station in available[:10]:
print(
station["站點名稱"],
station["可借車輛"],
station["更新時間"],
)
if __name__ == "__main__":
main()十四、來源網址不可任由使用者輸入
伺服器端程式若直接下載使用者提供的任意網址,可能造成SSRF,連到內部管理服務或雲端中繼資料。教學程式應使用固定官方端點或嚴格允許清單;不要關閉TLS憑證驗證,也不要在網址或日誌中放API金鑰。
十五、常見錯誤檢查表
| 現象 | 原因 | 修正 |
|---|---|---|
| 程式一直等待 | 沒有設定timeout | 明確設定連線與讀取逾時 |
| 404頁面被當JSON | 未先檢查HTTP狀態 | 呼叫raise_for_status() |
| json()成功後仍KeyError | 欄位改變或缺漏 | 驗證最外層及必要欄位 |
| 資料服務壓力過大 | 頻繁輪詢或無限重試 | 快取、退避、有限次數並遵守條款 |
| 舊資料看似即時 | 未保存或顯示擷取時間 | 一併保存來源與UTC時間 |
十六、USR實作挑戰
下載一次官方快照,統計有效筆數、錯誤筆數、行政區數量與更新時間範圍,不進行高頻輪詢。
保存一份符合授權的教學快照,讓沒有網路時仍可練習解析;畫面清楚標示「非即時資料」。
把站點欄位驗證概念改寫為智慧養殖感測API,設計來源、時間、有效筆數、錯誤筆數與過期警示。
十七、與生成式AI協作
請擔任Python requests與開放資料安全助教。
請檢查我的JSON下載程式:
1. 使用固定HTTPS官方端點;
2. 必須設定連線與讀取timeout;
3. 先raise_for_status()再解析JSON;
4. 驗證最外層型別及必要欄位;
5. 單筆錯誤不能拖垮整批;
6. 重試只限GET及特定暫時性錯誤;
7. 保存來源、擷取時間與資料品質;
8. 不關閉TLS驗證,不接受任意使用者網址。
請提供測試案例與失敗時的安全行為。下載JSON只需幾行,可靠使用卻需要逾時、狀態碼、格式、欄位、重試、大小與來源治理。網路資料永遠是不穩定的外部輸入,先驗證再使用。
沒有留言:
張貼留言