Тело ввода/вывода журнала AsyncClient

Можно ли расширить и параметризовать AsyncClient для регистрации тела ввода и вывода каждого внешнего вызова для повторного использования в нескольких методах, использующих один и тот же AsyncClient?

import json
import logging

from fastapi import FastAPI, HTTPException
from httpx import AsyncClient

# Set up basic configuration for logging
logging.basicConfig(level=logging.DEBUG)

app = FastAPI()
client = AsyncClient()


@app.get("/example_post")
async def example_post():
    url = "https://jsonplaceholder.typicode.com/posts"  # Free fake and reliable API for testing and prototyping.
    payload = {
        "title": 'fooxxx',
        "body": 'bar',
        "userId": 1,
    }

    # Perform the HTTP POST request; would like to log input payload and output data response
    response = await client.post(url, content=json.dumps(payload))

    # Parse response as JSON
    data = response.json()

    # Searching for universal solution that can log request and response of every api call instead of logging manually
    logging.info(f"Received response: {data}")

    return data


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host = "0.0.0.0", port=8000)

@Chris отметил этот вопрос как повторяющийся, и это не та же проблема. Мне нужно регистрировать ввод и вывод AsyncClient не по запросу на моей конечной точке (в данном случае example_post), а по URL-адресу.

Levijatanu 18.04.2024 18:27
Почему в Python есть оператор "pass"?
Почему в Python есть оператор "pass"?
Оператор pass в Python - это простая концепция, которую могут быстро освоить даже новички без опыта программирования.
Некоторые методы, о которых вы не знали, что они существуют в Python
Некоторые методы, о которых вы не знали, что они существуют в Python
Python - самый известный и самый простой в изучении язык в наши дни. Имея широкий спектр применения в области машинного обучения, Data Science,...
Основы Python Часть I
Основы Python Часть I
Вы когда-нибудь задумывались, почему в программах на Python вы видите приведенный ниже код?
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
Алиса и Боб имеют неориентированный граф из n узлов и трех типов ребер:
Оптимизация кода с помощью тернарного оператора Python
Оптимизация кода с помощью тернарного оператора Python
И последнее, что мы хотели бы показать вам, прежде чем двигаться дальше, это
Советы по эффективной веб-разработке с помощью Python
Советы по эффективной веб-разработке с помощью Python
Как веб-разработчик, Python может стать мощным инструментом для создания эффективных и масштабируемых веб-приложений.
1
1
97
1
Перейти к ответу Данный вопрос помечен как решенный

Ответы 1

Ответ принят как подходящий

Тело запроса

Когда дело доходит до регистрации тела запроса, вам следует учитывать перехватчики событий.

Вот документ: https://www.python-httpx.org/advanced/event-hooks/.

И вот как это можно реализовать:

from httpx import AsyncClient
from httpx import Response, Request

async def log_request(request: Request):
    print(f"Request event hook: {request.method} {request.url} {request.content} - Waiting for response")

client = AsyncClient(event_hooks = {'request': [log_request]})

Тело ответа

Что касается тела ответа, здесь все немного сложнее, поскольку тело представляет собой поток, который можно потребить только один раз.

Чтобы обойти эту проблему, вы можете написать собственный транспортный уровень. Вы можете обратиться к этому ответу: https://github.com/encode/httpx/discussions/3073#discussioncomment-8263619

Другие вопросы по теме