При работе с различными API-запросами, чаще всего на ум приходит библиотека requests. Она отлично выполняет свою функцию, однако, она не поддерживает асинхронное выполнение. На замену requests пришла новая библиотека httpx.
Пример выполнения GET-запроса:
response = httpx.get("https://reqres.in/api/users/2")
print(response.status_code)
print(response.json())
В этом примере выполнен базовый GET-запрос. Метод .get() возвращает объект Response, который содержит статус ответа, заголовки и данные ответа.
Пример выполнения асинхронного POST-запроса:
async def fetch_data():
async with httpx.AsyncClient() as client:
response = await client.post(
"https://reqres.in/api/users", data={"name": "Napkin", "job": "Author"}
)
print(response.status_code)
print(response.json())
asyncio.run(fetch_data())
В этом примере мы открыли асинхронный клиент в контекстном менеджере для выполнения запроса. В метод .post(), также передали аргумент data, содержащий словарь данных.
Пример выполнения stream-запроса:
async def fetch_data():
async with httpx.AsyncClient() as client:
async with client.stream(
"POST",
"https://api.openai.com/v1/chat/completions",
json={
"model": "gpt-4o-mini",
"messages": [
{"role": "system", "content": "You are a assistant."},
{"role": "user", "content": "Hello!"},
],
"stream": True,
},
headers={
"Authorization": "Bearer <token>"
},
) as response:
async for chunk in response.aiter_text():
yield chunk
async def main():
async for data in fetch_data():
print(data)
asyncio.run(main())
Этот пример показывает, как httpx может использоваться для выполнения запросов в режиме "стриминга".