import urllib.request

import urllib.request
import json
url = "https://api.example.com/data"
req = urllib.request.Request(url, method="GET")
with urllib.request.urlopen(req) as response:
    data = response.read()
    print(json.loads(data))

import urllib.parse
data = urllib.parse.urlencode({"key": "value"}).encode()
req = urllib.request.Request(url, data=data, method="POST")

pip install requests

import requests
resp = requests.get("https://api.example.com/data")
print(resp.status_code)
print(resp.json())

resp = requests.post(
    "https://api.example.com/login",
    json={"username": "pan", "password": "123"},
    timeout=5
)

session = requests.Session()
session.headers.update({
    "Authorization": "Bearer token"
})
resp = session.get("https://api.example.com/data")

requests.get(url, timeout=(3, 10))  # (connect, read)

try:
    resp = requests.get(url, timeout=5)
    resp.raise_for_status()
except requests.exceptions.Timeout:
    ...
except requests.exceptions.RequestException:
    ...

pip install httpx

import httpx
with httpx.Client(timeout=5.0) as client:
    resp = client.get("https://api.example.com/data")
    print(resp.json())

import httpx
import asyncio
async def main():
    async with httpx.AsyncClient(timeout=5.0) as client:
        resp = await client.get("https://api.example.com/data")
        print(resp.json())
asyncio.run(main())

async def handler():
    async with httpx.AsyncClient() as client:
        await client.get(url)

client = httpx.AsyncClient(timeout=5.0)
async def handler():
    resp = await client.get(url)

limits = httpx.Limits(
    max_connections=100,
    max_keepalive_connections=20
)
client = httpx.AsyncClient(limits=limits)

timeout = httpx.Timeout(
    connect=3.0,
    read=10.0,
    write=5.0,
    pool=3.0
)

from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3))
async def fetch():
    return await client.get(url)

class HttpClient:
    def __init__(self):
        self.client = httpx.AsyncClient(timeout=5.0)
    async def get(self, url, **kwargs):
        resp = await self.client.get(url, **kwargs)
        resp.raise_for_status()
        return resp.json()

pip install aiohttp

import aiohttp
import asyncio
async def main():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://api.example.com") as resp:
            print(await resp.json())
asyncio.run(main())

httpx.get(...)

from infra.http import http_client
await http_client.get(...)

start = time.perf_counter()
resp = await client.get(url)
cost = time.perf_counter() - start

requests

httpx + 連接池 + 重試

httpx.AsyncClient

aiohttp

httpx.AsyncClient

httpx.Client