HTTPX là một thư viện mạnh mẽ mới cho việc kết nối HTTP trong Python. Nó đang trở thành lựa chọn phổ biến nhất trong web scraping vì nó cung cấp khả năng kết nối không đồng bộ và hỗ trợ http2.
Trong bài viết này, chúng ta sẽ tìm hiểu về những điểm nổi bật của thư viện httpx trong việc web scraping và cách sử dụng nó một cách hiệu quả.
Cài đặt httpx
HTTPX là một gói Python thuần và nên được cài đặt dễ dàng bằng lệnh pip:
$ pip install httpxNgoài ra, nó cũng có thể được cài đặt bằng công cụ quản lý gói poetry:
$ poetry init -d httpx # hoặc $ poetry add httpxSử dụng HTTPX
HTTPX có thể được sử dụng cho các yêu cầu cá nhân trực tiếp và hỗ trợ hầu hết các chức năng HTTP phổ biến như yêu cầu GET, POST và có thể trích xuất dữ liệu JSON trực tiếp thành từ điển Python:
import httpx
# Yêu cầu GET
response = httpx.get("https://httpbin.dev/get")
print(response)
data = response.json()
print(data['url'])
# Yêu cầu POST
payload = {"query": "foo"}
# Dữ liệu content dạng application/json:
response = httpx.post("https://httpbin.dev/post", json=payload)
# Hoặc dữ liệu content dạng formdata:
response = httpx.post("https://httpbin.dev/post", data=payload)
print(response)
data = response.json()
print(data['url'])Ở đây, chúng ta sử dụng httpx để tải dữ liệu JSON bằng cách sử dụng phương thức .json() của đối tượng response. Httpx đi kèm với nhiều phím tắt tiện lợi và dễ dùng như thế này, làm cho nó trở thành một thư viện HTTP rất tiện lợi cho web scraping.
Sử dụng Client trong httpx
Đối với web scraping, tốt nhất là sử dụng một httpx.Client để áp dụng các thiết lập tùy chỉnh như tiêu đề, cookies và proxy cho toàn bộ phiên httpx:
import httpx
with httpx.Client(
# Bật hỗ trợ HTTP2
http2=True,
# Thiết lập tiêu đề cho tất cả các yêu cầu
headers={"x-secret": "foo"},
# Thiết lập cookies
cookies={"language": "en"},
# Thiết lập proxy
proxies={
# Thiết lập proxy cho tất cả các kết nối http://
"http": "http://222.1.1.1:8000",
# Thiết lập proxy cho tất cả các kết nối https://
"https": "http://222.1.1.1:8000",
# Cũng có thể sử dụng socks5, socks4 và socks4a
"https": "socks5://222.1.1.1:8000",
}
) as session:
# Gửi các yêu cầu
# ...httpx.Client áp dụng một bộ cấu hình cho tất cả các yêu cầu và even giữ track cookies được thiết lập bởi máy chủ.
Sử dụng httpx không đồng bộ
Để sử dụng httpx không đồng bộ với asyncio trong Python, bạn có thể sử dụng đối tượng httpx.AsyncClient():
import asyncio
import httpx
async def main():
async with httpx.AsyncClient(
# Giới hạn số kết nối không đồng bộ
limits=httpx.Limits(max_connections=10),
# Tăng timeout cho kết nối không đồng bộ
timeout=httpx.Timeout(60.0), # giây
# Lưu ý: AsyncClient nhận các đối số giống như Client (như tiêu đề, cookies, v.v.)
) as client:
# Gửi yêu cầu không đồng bộ
urls = [
"https://httpbin.dev/get",
"https://httpbin.dev/get",
"https://httpbin.dev/get",
]
responses = asyncio.gather(*[client.get(url) for url in urls])
# Hoặc sử dụng asyncio.as_completed:
for result in asyncio.as_completed([client.get(url) for url in urls]):
response = await result
print(response)
asyncio.run(main())Lưu ý rằng khi sử dụng async with, tất cả các kết nối phải hoàn thành trước khi đóng async with nếu không sẽ có lỗi:
RuntimeError: Cannot send a request, as the client has been closed.Thay vì async with statement, bạn cũng có thể mở/đóng thủ công httpx AsyncClient:
import asyncio
import httpx
async def main():
client = httpx.AsyncClient()
# ...
# Đóng client
await client.aclose()
asyncio.run(main())Xử lý sự cố trong HTTPX
Mặc dù httpx là một thư viện tuyệt vời, nhưng dễ gặp một số vấn đề phổ biến. Dưới đây là một số vấn đề phổ biến mà bạn có thể gặp phải khi web scraping bằng httpx và cách giải quyết chúng:
httpx.TimeoutException
Lỗi httpx.TimeoutException xảy ra khi yêu cầu mất nhiều thời gian hơn thời gian chờ ổn định/mặc định. Hãy thử tăng tham số timeout:
httpx.get("https://httpbin.org/delay/10", timeout=httpx.Timeout(60.0))httpx.ConnectError
Ngoại lệ httpx.ConnectError được ném khi xảy ra vấn đề kết nối, có thể do:
- Kết nối internet không ổn định.
- Máy chủ không thể truy cập.
- Sai sót trong tham số URL.
httpx.TooManyRedirects
Lỗi httpx.TooManyRedirects được ném khi một yêu cầu vượt quá số lượng tối đa của các chuyển hướng được phép.
Điều này có thể do vấn đề với máy chủ web bị scrape hoặc logic chuyển hướng của httpx. Có thể khắc phục bằng cách giải quyết các chuyển hướng thủ công:
response = httpx.get(
"https://httpbin.dev/redirect/3",
allow_redirects=False, # Vô hiệu hóa xử lý chuyển hướng tự động
)
# Sau đó, chúng ta có thể kiểm tra xem có muốn xử lý chuyển hướng tự động mình không:
redirect_location = response.headers["Location"]httpx.HTTPStatusError
Lỗi httpx.HTTPStatusError được ném khi sử dụng raise_for_status=True và mã trạng thái phản hồi từ máy chủ không nằm trong khoảng 200-299 như là 404:
response = httpx.get(
"https://httpbin.dev/redirect/3",
raise_for_status=True,
)
# Khi web scraping, mã trạng thái không nằm trong khoảng 200-299 có thể đồng nghĩa với việc scraper bị chặn.httpx.UnsupportedProtocol
Lỗi httpx.UnsupportedProtocol được ném khi URL được cung cấp trong giao thức bị thiếu hoặc không thuộc phạm vi http://, https://, file://, hoặc ftp://. Đây là lỗi thường gặp khi URL thiếu phần “https://”.
Thử lại các yêu cầu HTTPX
HTTPX không đi kèm với bất kỳ tính năng thử lại nào, nhưng nó có thể dễ dàng tích hợp với các gói thử lại phổ biến trong Python như tenacity (pip install tenacity).
Sử dụng tenacity, chúng ta có thể thêm logic thử lại cho các mã trạng thái nằm ngoài khoảng 200-299, các ngoại lệ httpx và thậm chí kiểm tra phần nội dung phản hồi để tìm kiếm từ khóa lỗi:
import httpx
from tenacity import retry, stop_after_attempt, wait_fixed, retry_if_exception_type, retry_if_result
# Định nghĩa điều kiện để thử lại dựa trên các loại ngoại lệ
def is_retryable_exception(exception):
return isinstance(exception, (httpx.TimeoutException, httpx.ConnectError))
# Định nghĩa điều kiện để thử lại dựa trên mã trạng thái HTTP
def is_retryable_status_code(response):
return response.status_code in [500, 502, 503, 504]
# Định nghĩa điều kiện để thử lại dựa trên nội dung phản hồi
def is_retryable_content(response):
return "you are blocked" in response.text.lower()
# Sử dụng decorator retry và xác định các quy tắc thử lại
@retry(
# Thử lại với các ngoại lệ
retry=(retry_if_exception_type(is_retryable_exception) |
retry_if_result(is_retryable_status_code) |
retry_if_result(is_retryable_content)),
# Thử lại tối đa 3 lần
stop=stop_after_attempt(3),
# Đợi cố định 5 giây giữa mỗi lần thử lại
wait=wait_fixed(5)
)
def fetch_url(url):
try:
response = httpx.get(url)
response.raise_for_status()
return response
except httpx.RequestError as e:
print(f"Request error: {e}")
raise e
url = "https://httpbin.org/get"
try:
response = fetch_url(url)
print(f"Successfully fetched URL: {url}")
print(response.text)
except Exception as e:
print(f"Failed to fetch URL: {url}")
print(f"Error: {e}")Trên đây là một ví dụ ngắn về cách áp dụng logic thử lại có thể xoay proxy và chuỗi User-Agent trong mỗi lần thử lại.
Trước tiên, chúng ta định nghĩa các pool proxy và pool User-Agent của mình, sau đó sử dụng decorator @retry để bao bọc hàm cào với logic thử lại của tenacity.
Để sửa đổi mỗi lần thử lại, chúng ta sử dụng tham số before_sleep, cho phép cập nhật cuộc gọi hàm cào của chúng ta với các tham số mới trong mỗi lần thử lại.
Dưới đây là một ví dụ chạy thử:
import httpx
import random
from tenacity import retry, stop_after_attempt, wait_random, retry_if_result
import asyncio
PROXY_POOL = [
"http://2.56.119.93:5074",
"http://185.199.229.156:7492",
"http://185.199.228.220:7300",
"http://185.199.231.45:8382",
"http://188.74.210.207:6286",
"http://188.74.183.10:8279",
"http://188.74.210.21:6100",
"http://45.155.68.129:8133",
"http://154.95.36.199:6893",
"http://45.94.47.66:8110",
]
USER_AGENT_POOL = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:54.0) Gecko/20100101 Firefox/54.0",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_6) AppleWebKit/604.3.5 (KHTML, like Gecko) Version/11.0.1 Safari/604.3.5",
]
# Định nghĩa điều kiện để thử lại dựa trên mã trạng thái HTTP
def is_retryable_status_code(response):
return response.status_code in [403, 404]
# Hàm gọi lại để sửa đổi cào sau mỗi lần thử lại
def update_scrape_call(retry_state):
# Thay đổi proxy ngẫu nhiên sau mỗi lần thử lại
new_proxy = random.choice(PROXY_POOL)
new_user_agent = random.choice(USER_AGENT_POOL)
print(
f"retry {retry_state.attempt_number}: {url} @ {proxy} with a new proxy {new_proxy}".format(
attempt_number=retry_state.attempt_number,
new_proxy=new_proxy,
**retry_state.kwargs
)
)
retry_state.kwargs["proxy"] = new_proxy
retry_state.kwargs["client_kwargs"]["headers"]["User-Agent"] = new_user_agent
@retry(
# Thử lại trên mã trạng thái không hợp lệ
retry=retry_if_result(is_retryable_status_code),
# Tối đa 5 lần thử lại
stop=stop_after_attempt(5),
# Đợi ngẫu nhiên 1-5 giây giữa mỗi lần thử lại
wait=wait_random(min=1, max=5),
# Cập nhật cào sau mỗi lần thử lại
before_sleep=update_scrape_call,
)
async def scrape(url, proxy, **client_kwargs):
async with httpx.AsyncClient(
proxies={"http://": proxy, "https://": proxy},
**client_kwargs,
) as client:
response = await client.get(url)
return response
# Đây là một demo ngắn về cách áp dụng logic thử lại có thể xoay proxy và chuỗi User-Agent trong mỗi lần thử lại.
async def example_run():
urls = [
"https://httpbin.dev/ip",
"https://httpbin.dev/ip",
"https://httpbin.dev/ip",
"https://httpbin.dev/status/403",
]
to_scrape = [
scrape(url=url, proxy=random.choice(PROXY_POOL), headers={"User-Agent": "foo"})
for url in urls
]
for result in asyncio.as_completed(to_scrape):
response = await result
print(response.json())
asyncio.run(example_run())Tránh bị chặn với Scrapfly
Scrapfly API cung cấp một SDK Python, tương tự như httpx nhưng mạnh mẽ hơn.
Scrapfly cung cấp các tính năng mạnh mẽ như:
- Bỏ qua bảo vệ chống scraping tự động
- Triệu hồi hàng triệu proxy cư trú từ hơn 50 quốc gia
- Trình duyệt headless để scrape các trang web động
- Bảng điều khiển để giám sát hiệu suất scrape
- Hỗ trợ Scrapy
Mọi chức năng của httpx đều được hỗ trợ bởi Scrapfly SDK, giúp việc chuyển đổi dễ dàng:
from scrapfly import ScrapeConfig, ScrapflyClient
client = ScrapflyClient(key="YOUR SCRAPFLY KEY")
result = client.scrape(
ScrapeConfig(
url="https://httpbin.dev/get",
# Kích hoạt bảo vệ chống scraping tự động (như cloudflare hoặc perimeterx)
bypass_asp=True,
# Chọn quốc gia proxy
country="US",
# Kích hoạt trình duyệt headless
render_js=True,
)
)
print(result.content)
# Mẹo: Sử dụng scrape đồng thời để tăng tốc độ scrape:
to_scrape = [
ScrapeConfig(url="https://httpbin.dev/get") for i in range(10)
]
async for result in client.concurrent_scrape(to_scrape):
print(result.content)SDK Scrapfly có thể được cài đặt bằng lệnh pip và miễn phí để thử: pip install scrapfly-sdk.
Câu hỏi thường gặp
Để kết thúc bài giới thiệu về httpx trong Python, hãy xem một số câu hỏi thường gặp liên quan đến web scraping với httpx.
So sánh giữa HTTPX và Requests
Requests là thư viện HTTP phổ biến nhất cho Python, nổi tiếng với việc dễ sử dụng và tiếp cận. Nó cũng là nguồn cảm hứng cho HTTPX, là người kế nhiệm tự nhiên của requests với tính năng Python hiện đại như hỗ trợ asyncio và http2.
So sánh giữa HTTPX và Aiohttp
Aiohttp là một trong những thư viện HTTP đầu tiên hỗ trợ asyncio và là một trong những nguồn cảm hứng cho HTTPX. Hai gói này rất tương tự nhau nhưng aiohttp đã được phát triển từ lâu hơn trong khi httpx mới hơn nhưng cung cấp nhiều tính năng hơn. Vì vậy, khi so sánh aiohttp và httpx cho web scraping, httpx được ưu tiên vì hỗ trợ http2.
Làm thế nào để sử dụng HTTP2 với HTTPX?
HTTPX hỗ trợ phiên bản http2, được khuyến nghị cho web scraping vì nó có thể giảm đáng kể tỷ lệ chặn của scraper. HTTP2 không được bật theo mặc định và vì vậy, cần sử dụng tham số http2=True trong đối tượng httpx.Client(http2=True) và httpx.AsyncClient(http2=True).
Làm thế nào để theo dõi tự động chuyển hướng trong HTTPX?
HTTPX không theo dõi chuyển hướng theo mặc định giống như các thư viện Python khác như requests. Để bật chuyển hướng tự động, hãy sử dụng tham số allow_redirects=True trong các phương thức yêu cầu httpx như httpx.get(url, allow_redirects=True) hoặc các đối tượng httpx client như httpx.Client(allow_redirects=True).
Tóm lược
HTTPX là một thư viện HTTP xuất sắc đã trở thành tiêu chuẩn de facto trong cộng đồng web scraping Python. Nó cung cấp các tính năng như hỗ trợ http2 và asyncio, giảm rủi ro bị chặn và cho phép web scraping đồng thời.
Kết hợp với tenacity, httpx giúp việc yêu cầu tài nguyên web trở nên dễ dàng với logic thử lại mạnh mẽ như xoay proxy và tiêu đề User-Agent.
Đây là một bài viết mẫu về cách sử dụng HTTPX và Python để lấy dữ liệu từ trang web. HTTPX là một công cụ mạnh mẽ cho việc web scraping và tạo ra nhiều khả năng và tính linh hoạt trong quá trình này. Bạn có thể tận dụng những lợi ích mà HTTPX mang lại và trải nghiệm sức mạnh của nó trong quá trình làm web scraping.
Conclusion: So above is the Cách sử dụng HTTPX và Python để Lấy dữ liệu từ trang web article. Hopefully with this article you can help you in life, always follow and read our good articles on the website: natuts.com
