🐍 Ngày 40 - Python hằng ngày 365 ngày - Tương tác với dữ liệu từ API
Chủ đề hôm nay sẽ giúp bạn lấy dữ liệu thời gian thực từ API, ví dụ như giá tiền ảo từ CoinGecko hoặc thời tiết từ OpenWeather. Đây là kỹ năng cực kỳ quan trọng khi bạn muốn phân tích hoặc tự động hóa quy trình lấy dữ liệu từ các dịch vụ online.
🧠 Kiến thức chính
API là gì?
API (Application Programming Interface) là giao diện cho phép ứng dụng của bạn giao tiếp với hệ thống khác thông qua các lệnh HTTP như GET, POST...
Các bước để tương tác với API trong Python:
- Dùng thư viện
requestsđể gửi yêu cầu tới API. - Phân tích dữ liệu nhận được (thường là JSON).
- Xử lý và hiển thị dữ liệu theo yêu cầu.
✏️ Ví dụ thực tế: Lấy giá Bitcoin từ CoinGecko API
import requests
url = "https://api.coingecko.com/api/v3/simple/price?ids=bitcoin&vs_currencies=usd"
response = requests.get(url)
data = response.json()
print(f"Giá Bitcoin hiện tại là: ${data['bitcoin']['usd']}")
🎯 Bài tập thực hành
Bài 1: Lấy dữ liệu thời tiết từ OpenWeather
- Tạo tài khoản tại https://openweathermap.org/api và lấy API key miễn phí.
- Lấy thông tin thời tiết hiện tại ở Hà Nội và in ra màn hình.
Bài 2: Gọi dữ liệu giá ETH từ CoinGecko
- Tương tự như ví dụ trên, nhưng với Ethereum (
eth).
✅ Ghi chú
- Một số API cần có
API key, bạn phải đăng ký tài khoản để lấy key. - Dữ liệu thường trả về là JSON, cần dùng
.json()để chuyển đổi.
Bạn có thể mở bài học này trên Google Colab để chạy trực tiếp:
🎥 Video hướng dẫn
🌐 API là gì và hoạt động ra sao
API (Application Programming Interface) là cách hai chương trình nói chuyện với nhau. Web API dùng HTTP — cùng giao thức mà trình duyệt dùng để mở trang web.
Client (Python) Server (Binance)
| |
|------ GET /api/v3/ticker/price --------->|
| (request) |
| |-- xử lý
|<----- 200 OK + JSON --------------------|
| (response) |
Bảng HTTP method cần nắm
| Method | Mục đích | Idempotent | Có body? |
|---|---|---|---|
GET | Đọc dữ liệu | ✅ | Không |
POST | Tạo mới / gửi dữ liệu | ❌ | Có |
PUT | Thay thế | ✅ | Có |
DELETE | Xoá | ✅ | Không |
Idempotent nghĩa là gọi 10 lần cho kết quả giống gọi 1 lần. Điều này quan trọng khi request bị mất mạng — gửi lại GET an toàn, gửi lại POST có thể tạo dữ liệu trùng.
Mã trạng thái — đọc để biết chuyện gì xảy ra
| Nhóm | Ý nghĩa | Ví dụ |
|---|---|---|
| 2xx | Thành công | 200 OK, 201 Created |
| 4xx | Lỗi phía bạn | 400 dữ liệu sai, 401 chưa đăng nhập, 403 không đủ quyền, 404 không tìm thấy, 429 gọi quá nhanh |
| 5xx | Lỗi phía server | 500 bug server, 503 quá tải |
⚠️ Nguyên tắc debugging API số 1:
4xxlà lỗi của bạn (sai tham số, thiếu key, gọi quá nhanh).5xxlà lỗi của server (thử lại sau). Đọc mã trạng thái trước khi đọc nội dung.
🐍 Ba cách gọi API trong Python
# CÁCH 1: urllib — có sẵn trong Python, không cần cài
import urllib.request
import json
with urllib.request.urlopen("https://api.binance.com/api/v3/ticker/price?symbol=BTCUSDT", timeout=10) as r:
du_lieu = json.loads(r.read().decode("utf-8"))
print(f"BTC: {float(du_lieu['price']):,.2f} USDT")
# CÁCH 2: requests — thư viện phổ biến nhất, dễ dùng hơn nhiều
# pip install requests
import requests
phan_hoi = requests.get(
"https://api.binance.com/api/v3/ticker/price",
params={"symbol": "BTCUSDT"},
timeout=10,
)
phan_hoi.raise_for_status() # ném lỗi nếu 4xx/5xx
print(phan_hoi.json())
# CÁCH 3: httpx — hỗ trợ async, dùng khi cần gọi nhiều API cùng lúc
# pip install httpx
import httpx
with httpx.Client(timeout=10) as client:
r = client.get("https://api.binance.com/api/v3/ticker/price",
params={"symbol": "ETHUSDT"})
print(r.json())
| Cách | Ưu điểm | Nhược điểm |
|---|---|---|
urllib | Không cần cài gì | Cú pháp dài dòng |
requests | Dễ nhất, nhiều tính năng | Không hỗ trợ async |
httpx | Hỗ trợ async, API giống requests | Cần cài thêm |
🛡️ Gọi API an toàn — hàm chuẩn dùng được mọi dự án
import time
import logging
import requests
logger = logging.getLogger(__name__)
def goi_api_an_toan(
url: str,
params: dict | None = None,
so_lan_thu: int = 3,
timeout: int = 15,
) -> dict | list | None:
"""Gọi API với retry, timeout, và log đầy đủ."""
cho = 1.0
for lan in range(1, so_lan_thu + 1):
try:
r = requests.get(url, params=params, timeout=timeout)
# 429 = quá nhiều request → chờ theo header Retry-After
if r.status_code == 429:
cho_theo_header = int(r.headers.get("Retry-After", cho))
logger.warning("Bị giới hạn (429), chờ %ss", cho_theo_header)
time.sleep(cho_theo_header)
continue
# 5xx = lỗi server → thử lại
if r.status_code >= 500:
logger.warning("Lỗi server %s, thử lại sau %ss", r.status_code, cho)
time.sleep(cho)
cho *= 2
continue
# 4xx khác = lỗi của mình, không nên thử lại
if 400 <= r.status_code < 500:
logger.error("Lỗi yêu cầu %s: %s", r.status_code, r.text[:200])
return None
return r.json()
except requests.Timeout:
logger.warning("Hết thời gian chờ (lần %s/%s)", lan, so_lan_thu)
except requests.ConnectionError:
logger.warning("Lỗi kết nối (lần %s/%s)", lan, so_lan_thu)
except ValueError as e:
logger.error("Response không phải JSON: %s", e)
return None
if lan < so_lan_thu:
time.sleep(cho)
cho *= 2
logger.error("Đã thử %s lần nhưng không thành công", so_lan_thu)
return None
du_lieu = goi_api_an_toan(
"https://api.binance.com/api/v3/ticker/24hr",
params={"symbol": "BTCUSDT"},
)
if du_lieu:
print(f"BTC 24h: {float(du_lieu['lastPrice']):,.2f} "
f"({float(du_lieu['priceChangePercent']):+.2f}%)")
Năm điểm quan trọng trong hàm trên:
| Điểm | Vì sao |
|---|---|
timeout=15 | Không có timeout, request có thể treo vô hạn |
| Retry với backoff cấp số nhân | Server đang quá tải cần thời gian hổi phục |
Đọc header Retry-After | Binance nói rõ bạn phải chờ bao lâu |
Không retry lỗi 4xx | Lỗi của mình thì thử lại cũng vô ích |
Bắt ValueError | Server có thể trả HTML (trang lỗi) thay vì JSON |
⚡ Gọi nhiều API cùng lúc — nhanh hơn nhiều lần
import concurrent.futures
import requests
CAP = ["BTCUSDT", "ETHUSDT", "SOLUSDT", "BNBUSDT", "XRPUSDT",
"ADAUSDT", "DOGEUSDT", "DOTUSDT", "MATICUSDT", "LTCUSDT"]
def lay_gia(cap: str) -> tuple[str, float | None]:
try:
r = requests.get(
"https://api.binance.com/api/v3/ticker/price",
params={"symbol": cap},
timeout=10,
)
return cap, float(r.json()["price"])
except Exception:
return cap, None
# Tuần tự: 10 request × ~0,2s = ~2 giây
# Đồng thời: ~0,3 giây
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
ket_qua = dict(executor.map(lay_gia, CAP))
for cap, gia in sorted(ket_qua.items(), key=lambda x: -(x[1] or 0)):
print(f"{cap:<12} {gia:>12,.4f}" if gia else f"{cap:<12} LỖI")
💡 Lưu ý: giới hạn
max_workers=5là quan trọng. Mở 50 luồng cùng lúc sẽ khiến Binance chặn IP của bạn ngay lập tức. Hãy tôn trọng rate limit bằng cách giới hạn số kết nối đồng thời.
📝 Bài tập mở rộng
Bài 1: Đọc API công khai bất kỳ
Thử gọi một trong các API miễn phí sau và in kết quả ra màn hình:
https://api.coingecko.com/api/v3/simple/price?ids=bitcoin&vs_currencies=usdhttps://api.open-meteo.com/v1/forecast?latitude=21.03&longitude=105.85¤t_weather=truehttps://api.github.com/users/torvalds
Bài 2: Lưu dữ liệu vào file
Gọi API giá 10 cặp, lưu kết quả kèm timestamp vào file JSON.
Bài 3: So sánh tuần tự vs đồng thời
Dùng time.perf_counter() để đo thời gian gọi 10 API theo hai cách. Chênh lệch bao nhiêu lần?
Bài 4: Xử lý lỗi
Cố tình gọi một cặp không tồn tại (symbol=KHONGCO) và xem API trả gì. Sửa code để xử lý trường hợp này mà không crash.
🎯 Tóm tắt ngày 40
✅ Hiểu API và giao thức HTTP
✅ Ba cách gọi API: urllib, requests, httpx
✅ Viết hàm gọi API có retry, timeout, xử lý 429/5xx
✅ Gọi nhiều API đồng thời bằng ThreadPoolExecutor
✅ Tôn trọng rate limit — giới hạn số kết nối đồng thời
Xem video minh họa chi tiết tại đây:
👉 YouTube - Làm việc với API Binance