Python Requests: современное HTTP в несколько строк кода

pip install requests

conda install -c anaconda requests

import requests

import requests

# Простой GET-запрос
response = requests.get('https://api.github.com/events')

# Проверка статуса ответа
if response.status_code == 200:
print('Успешный запрос!')
# Получаем содержимое ответа в формате JSON
data = response.json()
print(f"Получено {len(data)} событий")
else:
print(f"Ошибка: код {response.status_code}")

# Отправка данных в формате JSON
payload = {'key1': 'value1', 'key2': 'value2'}
response = requests.post('https://httpbin.org/post', json=payload)

# Отправка данных в формате form-data
response = requests.post('https://httpbin.org/post', data={'key': 'value'})

# PUT-запрос
response = requests.put('https://httpbin.org/put', data={'key': 'updated_value'})

# DELETE-запрос
response = requests.delete('https://httpbin.org/delete')

# PATCH-запрос (частичное обновление ресурса)
response = requests.patch('https://httpbin.org/patch', data={'key': 'partial_update'})

# HEAD-запрос (получение только заголовков)
response = requests.head('https://httpbin.org/get')

response = requests.get('https://api.github.com/events')

# Статус-код ответа
print(response.status_code) # 200

# Содержимое ответа в виде текста
print(response.text)

# Содержимое ответа в виде байтов
print(response.content)

# Автоматическое декодирование JSON
print(response.json())

# Заголовки ответа
print(response.headers)

# Кодировка ответа
print(response.encoding)

try:
response = requests.get('https://nonexistentwebsite.abc')
response.raise_for_status() # Вызовет исключение при статус-кодах 4xx/5xx
except requests.exceptions.RequestException as err:
print(f"Произошла ошибка: {err}")

# Способ 1: Передача параметров через params
payload = {'key1': 'value1', 'key2': 'value2'}
response = requests.get('https://httpbin.org/get', params=payload)
print(response.url) # https://httpbin.org/get?key1=value1&key2=value2

# Способ 2: Передача списка значений
payload = {'key1': 'value1', 'key2': ['value2', 'value3']}
response = requests.get('https://httpbin.org/get', params=payload)
print(response.url) # https://httpbin.org/get?key1=value1&key2=value2&key2=value3

headers = {
'User-Agent': 'Python Requests App',
'Accept': 'application/json',
'Content-Type': 'application/json'
}
response = requests.get('https://api.github.com/events', headers=headers)

# Использование кастомного User-Agent
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'}
response = requests.get('https://www.example.com', headers=headers)

files = {'file': ('report.csv', open('report.csv', 'rb'), 'text/csv')}
response = requests.post('https://httpbin.org/post', files=files)

# Отправка запроса с куки
cookies = {'session_id': '12345', 'user_id': '54321'}
response = requests.get('https://httpbin.org/cookies', cookies=cookies)

# Получение куки из ответа
cookies_received = response.cookies
print(cookies_received.get_dict())

# Установка таймаута (первое число – таймаут соединения, второе – таймаут чтения)
response = requests.get('https://api.github.com', timeout=(3.05, 27))

# Только общий таймаут
response = requests.get('https://api.github.com', timeout=5)

# Пример использования прокси
proxies = {
'http': 'http://10.10.1.10:3128',
'https': 'http://10.10.1.10:1080',
}
response = requests.get('https://example.org', proxies=proxies)

# Потоковая загрузка большого файла
response = requests.get('https://example.com/large_file.zip', stream=True)
with open('large_file.zip', 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)

# Базовая HTTP-аутентификация
response = requests.get('https://api.github.com/user', auth=('username', 'password'))

# Альтернативный способ
from requests.auth import HTTPBasicAuth
response = requests.get('https://api.github.com/user', auth=HTTPBasicAuth('username', 'password'))

# OAuth 2.0 аутентификация с использованием токена
headers = {'Authorization': 'Bearer YOUR_ACCESS_TOKEN'}
response = requests.get('https://api.service.com/endpoint', headers=headers)

# Использование OAuth 1.0 аутентификации
from requests_oauthlib import OAuth1
auth = OAuth1('YOUR_APP_KEY', 'YOUR_APP_SECRET',
'USER_OAUTH_TOKEN', 'USER_OAUTH_TOKEN_SECRET')
response = requests.get('https://api.twitter.com/1.1/account/verify_credentials.json', auth=auth)

# Обработка JSON-ответа
response = requests.get('https://api.github.com/events')
json_data = response.json()
for event in json_data:
print(f"Event ID: {event.get('id')}, Type: {event.get('type')}")

# Обработка XML-ответа
response = requests.get('https://www.w3schools.com/xml/note.xml')
import xml.etree.ElementTree as ET
root = ET.fromstring(response.content)
for child in root:
print(f"{child.tag}: {child.text}")

response = requests.get('https://api.github.com/repos/invalid/repo')

# Проверка статус-кода
if response.status_code == 200:
print("Успешный запрос")
elif response.status_code == 404:
print("Ресурс не найден")
elif response.status_code == 401:
print("Требуется аутентификация")
elif 500 <= response.status_code < 600:
print("Ошибка на стороне сервера")

# Обработка пагинации в API GitHub
all_events = []
url = 'https://api.github.com/events'

while url:
response = requests.get(url)
all_events.extend(response.json())

# Проверяем наличие ссылки на следующую страницу в заголовках
if 'next' in response.links.keys():
url = response.links['next']['url']
else:
url = None

print(f"Всего собрано {len(all_events)} событий")

import time

# Простая реализация с контролем скорости запросов
for i in range(10):
response = requests.get('https://api.github.com/events')

# Проверяем оставшееся количество запросов
rate_limit_remaining = int(response.headers.get('X-RateLimit-Remaining', 0))

if rate_limit_remaining < 10:
# Если мы почти достигли лимита, делаем паузу
reset_time = int(response.headers.get('X-RateLimit-Reset', 0))
sleep_time = max(reset_time – time.time(), 0) + 1
print(f"Приближаемся к лимиту запросов. Ожидание {sleep_time} секунд...")
time.sleep(sleep_time)

print(f"Запрос {i+1} выполнен, осталось {rate_limit_remaining} запросов")

# Добавляем небольшую задержку между запросами
time.sleep(0.5)

import requests

# Создаем сессию
session = requests.Session()

# Устанавливаем заголовки, которые будут использоваться во всех запросах
session.headers.update({'User-Agent': 'Mozilla/5.0', 'Accept': 'application/json'})

# Выполняем запросы в рамках сессии
response1 = session.get('https://httpbin.org/cookies/set/sessioncookie/123456789')
response2 = session.get('https://httpbin.org/cookies')

# Cookies автоматически сохраняются между запросами
print(response2.json()) # Должен содержать 'sessioncookie': '123456789'

# Закрываем сессию после использования
session.close()

# Без использования сессий
for i in range(10):
response = requests.get('https://api.github.com/events')

# С использованием сессий – намного эффективнее
session = requests.Session()
for i in range(10):
response = session.get('https://api.github.com/events')
session.close()

# Настройка пула соединений и повторных попыток
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Создаем стратегию повторных попыток
retry_strategy = Retry(
total=3, # максимальное количество повторных попыток
status_forcelist=[429, 500, 502, 503, 504], # коды, при которых повторяем запрос
backoff_factor=1, # фактор экспоненциальной задержки
respect_retry_after_header=True # учитывать заголовок Retry-After
)

# Создаем адаптер с нашей стратегией
adapter = HTTPAdapter(max_retries=retry_strategy, pool_connections=10, pool_maxsize=10)

# Создаем сессию и привязываем адаптер к определенным схемам URL
session = requests.Session()
session.mount("http://", adapter)
session.mount("https://", adapter)

# Теперь все запросы через эту сессию будут использовать настроенный адаптер
response = session.get('https://api.example.com', timeout=5)

# Эмуляция процесса входа в веб-приложение
session = requests.Session()

# Сначала получаем страницу логина, чтобы получить CSRF-токен
login_page = session.get('https://example.com/login')

# Здесь можно использовать библиотеку BeautifulSoup для парсинга HTML
# и извлечения CSRF-токена
import re
csrf_token = re.search('name="csrf_token" value="(.+?)"', login_page.text).group(1)

# Отправляем форму логина
login_data = {
'username': 'myuser',
'password': 'mypassword',
'csrf_token': csrf_token
}
login_response = session.post('https://example.com/login', data=login_data)

# Теперь сессия аутентифицирована, и мы можем получать защищенные страницы
protected_page = session.get('https://example.com/protected')
print('Успешный вход' if 'Welcome, myuser' in protected_page.text else 'Ошибка входа')

import concurrent.futures
import time

# Функция для выполнения запроса
def fetch_url(session, url):
start = time.time()
response = session.get(url)
return {
'url': url,
'status': response.status_code,
'time': time.time() – start
}

# Список URL для запроса
urls = [
'https://www.python.org',
'https://www.github.com',
'https://www.stackoverflow.com',
'https://www.wikipedia.org',
'https://www.reddit.com'
]

# Создаем сессию с нужными настройками
session = requests.Session()
session.headers.update({'User-Agent': 'Mozilla/5.0'})

# Выполняем запросы параллельно
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
# Создаем набор задач для выполнения
tasks = [executor.submit(fetch_url, session, url) for url in urls]

# Обрабатываем результаты по мере их завершения
for future in concurrent.futures.as_completed(tasks):
result = future.result()
print(f"URL: {result['url']}, Status: {result['status']}, Time: {result['time']:.2f}s")

from requests_toolbelt.sessions import SessionSerializer

# Создаем и настраиваем сессию
session = requests.Session()
session.get('https://httpbin.org/cookies/set/sessionid/12345')

# Сохраняем состояние сессии
serialized = SessionSerializer().dumps(session)
with open('session.pickle', 'wb') as f:
f.write(serialized)

# В другом месте или в другое время можно восстановить сессию
with open('session.pickle', 'rb') as f:
restored_session = SessionSerializer().loads(f.read())

# Продолжаем использовать восстановленную сессию
response = restored_session.get('https://httpbin.org/cookies')
print(response.json()) # Должен содержать сохраненные cookies