Python: полный мануал с примерами и справочником функций

Глава 21 · средний

Сеть, HTTP и взаимодействие с сервисом

Сделайте локальный HTTP-запрос, проверьте ответ и различайте ошибки транспорта, статуса и данных.

Собирайте параметры средствами библиотеки

HTTP-клиенту нужны адрес, метод, заголовки и правила обработки результата. Начните с адреса: путь и query-параметры — разные компоненты URL. urllib.parse.urlencode корректно кодирует пробелы, кириллицу и специальные символы в параметрах. Не склеивайте пользовательский текст через «?q=» и амперсанды: часть строки может превратиться в другой параметр. Для повторяющегося поля используйте список значений и doseq=True. Фрагмент после решётки предназначен для клиента и обычно не передаётся серверу в HTTP-запросе.

from urllib.parse import urlencode, urlsplit, parse_qs, urljoin

query = urlencode({"q": "CSS сетка", "tag": ["python", "web"]}, doseq=True)
url = "https://example.test/search?" + query
parts = urlsplit(url)
print("Хост:", parts.hostname)
print("Запрос:", parse_qs(parts.query)["q"][0])
print("Метки:", ", ".join(parse_qs(parts.query)["tag"]))
print(urljoin("https://example.test/docs/", "guide"))

Хост: example.test
Запрос: CSS сетка
Метки: python, web
https://example.test/docs/guide

urljoin разрешает относительный адрес, но абсолютный второй аргумент способен заменить исходный хост. Если программа допускает только определённый сервис, проверяйте схему и конечный адрес по выбранной политике, учитывая перенаправления. Парсер URL не является проверкой доступности и доверия к адресу. В этой главе example.test используется только для разбора строки; настоящие запросы направляются исключительно на собственный сервер 127.0.0.1. Внешний аккаунт, токен и установленный сетевой пакет для воспроизведения не нужны.

Для библиотеки удобно передавать настроенный клиент отдельным объектом. Глобальная установка opener меняет поведение других запросов в процессе; собственный экземпляр оставляет настройку рядом с кодом, которому она нужна, и упрощает подмену источника в тесте.

Получите ответ собственного учебного сервера

Следующий скрипт запускает сервер на свободном локальном порту, делает запрос и завершает сервер в finally. Номер порта не выводится, поэтому результат остаётся воспроизводимым. ProxyHandler({}) относится только к учебному localhost-клиенту и исключает влияние настроенного внешнего прокси. http.server удобен для такого стенда, но не заменяет производственный сервер с полноценной обработкой доступа и нагрузки. Не привязывайте учебный сервер к публичному интерфейсу без отдельной задачи: здесь он доступен только с этого компьютера.

import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Thread
from urllib.request import build_opener, ProxyHandler

class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        body = b'{"quantity": 3}'
        self.send_response(200)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)
    def log_message(self, format, *args):
        pass

server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
thread = Thread(target=server.serve_forever, daemon=True)
thread.start()
try:
    url = f"http://127.0.0.1:{server.server_port}/data"
    opener = build_opener(ProxyHandler({}))
    with opener.open(url, timeout=3) as response:
        body = response.read(1025)
        if len(body) > 1024:
            raise ValueError("Ответ слишком большой")
        data = json.loads(body.decode("utf-8"))
        if not isinstance(data, dict) or type(data.get("quantity")) is not int:
            raise ValueError("Неожиданная структура ответа")
        print("Статус:", response.status)
        print("Количество:", data["quantity"])
finally:
    server.shutdown()
    server.server_close()
    thread.join(timeout=3)

Статус: 200
Количество: 3

Статус 200 — ещё не правильные данные

Успешный HTTP-статус говорит о результате запроса на уровне протокола, а не о соответствии ответа вашей модели. Проверьте допустимый Content-Type, ограничьте размер, декодируйте известную кодировку и только затем разбирайте JSON. После разбора проверьте обязательные поля, типы и диапазоны. Заголовок Content-Length может отсутствовать или быть непригодным для доверенной оценки; ограничение чтения всё равно нужно. В примере читается максимум на один байт больше лимита, чтобы отличить разрешённый размер от превышения.

Ответ — ресурс, который следует закрывать через with. Таймаут urlopen относится к блокирующим операциям, например установлению соединения, и не обязательно ограничивает полное время всей задачи. При сложном клиенте отдельно задают общий дедлайн, время подключения и ожидания данных согласно возможностям библиотеки. Если ответ может сжиматься, ограничивайте и объём после распаковки. Зависание и чрезмерный ответ должны приводить к понятной ошибке, а не к бесконечному ожиданию или загрузке всей памяти.

Различайте HTTP-ошибку и ошибку JSON

HTTPError содержит код HTTP, а URLError описывает проблему доступа или транспорта. HTTPError является подклассом URLError, поэтому более конкретный обработчик ставят первым. Ошибка декодирования и JSONDecodeError относятся уже к содержимому ответа. Не объединяйте их сообщением «нет интернета»: сервер мог успешно вернуть неверные данные. При регистрации ошибки оставляйте полезную причину и идентификатор учебной операции, не печатая заголовок авторизации или полный документ пользователя.

import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Thread
from urllib.error import HTTPError
from urllib.request import build_opener, ProxyHandler

class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/missing":
            self.send_error(404)
            return
        body = b'{broken}'
        self.send_response(200)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)
    def log_message(self, format, *args):
        pass

server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
thread = Thread(target=server.serve_forever, daemon=True)
thread.start()
opener = build_opener(ProxyHandler({}))
try:
    for path in ["/missing", "/bad-json"]:
        try:
            with opener.open(f"http://127.0.0.1:{server.server_port}{path}", timeout=3) as response:
                json.loads(response.read(1024).decode("utf-8"))
        except HTTPError as error:
            print("HTTP:", error.code)
            error.close()
        except json.JSONDecodeError:
            print("Данные: неверный JSON")
finally:
    server.shutdown()
    server.server_close()
    thread.join(timeout=3)

HTTP: 404
Данные: неверный JSON

Метод определяет ожидаемое действие

GET обычно запрашивает представление ресурса, POST часто создаёт или запускает действие, PUT заменяет заданное представление, DELETE удаляет ресурс по контракту сервиса. Реальный смысл определяется API. Request позволяет явно задать method и headers; при отправке JSON сформируйте байты UTF-8 и соответствующий Content-Type. Не помещайте секретный токен в исходник, адрес или пример лога. Cookie, перенаправления и авторизация требуют отдельной политики; стандартная библиотека предоставляет инструменты, но не угадывает правила конкретного сервиса.

Повтор должен быть ограниченным и осмысленным

Не повторяйте любой неудачный запрос бесконечно. Ограничьте число попыток, выберите временные ошибки и паузу; при подходящем ответе учтите Retry-After. Идемпотентность означает, что повтор не должен создавать новый дополнительный эффект. Даже если клиент не получил ответ на создание записи, сервер мог уже выполнить действие. Для такого API нужна согласованная защита от дублей, например поддерживаемый сервисом ключ идемпотентности. Ошибку схемы JSON повтор обычно не исправляет; сначала отделите сетевую попытку от преобразования данных.

from urllib.error import URLError

attempts = 0
def read_demo():
    global attempts
    attempts += 1
    if attempts < 3:
        raise URLError("временный учебный сбой")
    return {"count": 4}

def retry_read(operation, limit):
    for number in range(1, limit + 1):
        try:
            return operation()
        except URLError:
            if number == limit:
                raise
    raise ValueError("Нужна хотя бы одна попытка")

data = retry_read(read_demo, 3)
print("Попыток:", attempts)
print("Получено:", data["count"])

Попыток: 3
Получено: 4

HTTP-клиент, TLS и сокет — разные уровни

socket работает с низкоуровневым обменом байтами: одно чтение не обязано вернуть целое сообщение. Для HTTP используйте клиент, который понимает протокол, вместо ручной склейки сетевых строк. HTTPS добавляет TLS; стандартный безопасный контекст проверяет сертификат и имя хоста. Не отключайте эту проверку как обычное исправление подключения: сначала проверьте адрес, время системы и доверенные сертификаты. Сторонний клиент или веб-фреймворк выбирают по требованиям задачи, явно отделяя его от стандартной библиотеки.

Практика: проверка ответа без настоящей сети

Выделите преобразование ответа в отдельную функцию. Пусть она принимает байты с JSON и возвращает неотрицательное целое count. Проверьте обычный ответ, логическое значение и повреждённый JSON. Это тест границы данных, поэтому запуск сервера для него лишний. После такой проверки настоящий HTTP-слой может использовать ту же функцию, сохраняя разные сообщения о транспортной ошибке и неправильном содержимом.

import json

def parse_response(body):
    value = json.loads(body.decode("utf-8"))
    if not isinstance(value, dict) or type(value.get("count")) is not int or value["count"] < 0:
        raise ValueError("Нужно неотрицательное целое count")
    return value["count"]

for body in [b'{"count": 2}', b'{"count": true}', b'not-json']:
    try:
        print(parse_response(body))
    except (ValueError, UnicodeDecodeError):
        print("Неверные данные")

2
Неверные данные
Неверные данные