Как в скрипте на Python прочитать тело (body) HTTP-запроса?

В DevOps-скриптах на Python часто нужно работать с HTTP-запросами, будь то вызов API или создание простого веб-сервиса для мониторинга. Вот как я это делаю.

1. Клиентская сторона (отправка запроса и чтение ответа): Использую библиотеку requests — де-факто стандарт.

import requests
import json

# Пример POST-запроса с JSON телом
api_url = "https://api.example.com/webhook"
payload = {"event": "deployment", "status": "success", "version": "1.2.3"}
headers = {"Content-Type": "application/json", "Authorization": "Bearer token123"}

try:
    response = requests.post(api_url, json=payload, headers=headers, timeout=10)
    response.raise_for_status()  # Выбросит исключение при HTTP-ошибке (4xx, 5xx)

    # Чтение тела ответа
    if response.headers.get('Content-Type') == 'application/json':
        body = response.json()  # Парсинг JSON
        print(f"ID задачи: {body.get('taskId')}")
    else:
        body = response.text    # Чтение как plain text
        print(body)

except requests.exceptions.RequestException as e:
    print(f"Ошибка запроса: {e}")
    # Логируем ошибку в Sentry/ELK

2. Серверная сторона (обработка входящих запросов): Для создания легковесных API, веб-хуков или health-check эндпоинтов в инфраструктурных скриптах удобен FastAPI (быстрый и с автоматической валидацией).

from fastapi import FastAPI, Request, HTTPException
import uvicorn

app = FastAPI()

@app.post("/webhook")
async def handle_webhook(request: Request):
    """Эндпоинт для приёма webhook от CI/CD системы (например, GitLab)."""
    # Проверяем заголовок для безопасности
    if request.headers.get("X-GitLab-Token") != "my-secret-token":
        raise HTTPException(status_code=403, detail="Forbidden")

    # Чтение тела запроса как JSON
    body = await request.json()
    event_type = body.get("object_kind")

    # Логика обработки
    if event_type == "pipeline":
        print(f"Pipeline status: {body.get('object_attributes', {}).get('status')}")
        # Здесь можно запустить ansible-playbook или обновить статус в Prometheus

    return {"status": "received"}

# Для запуска: uvicorn script_name:app --host 0.0.0.0 --port 8080

3. Низкоуровневая обработка (без внешних библиотек): Иногда в минималистичных окружениях можно использовать стандартную библиотеку http.server.

from http.server import HTTPServer, BaseHTTPRequestHandler
import json

class SimpleHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        content_length = int(self.headers.get('Content-Length', 0))
        # Чтение сырого тела запроса
        body_raw = self.rfile.read(content_length)
        # Попытка парсинга JSON
        try:
            body = json.loads(body_raw.decode('utf-8'))
        except json.JSONDecodeError:
            body = None

        self.send_response(200)
        self.end_headers()
        self.wfile.write(b"OK")
        # Далее работа с body...

# Запуск сервера
server = HTTPServer(('localhost', 8000), SimpleHandler)
server.serve_forever()

Ключевые моменты для DevOps: Всегда добавляю таймауты, обработку ошибок, логирование и, при необходимости, базовую аутентификацию в такие скрипты.