18804694760

Committed 25 Oct 2025 03:02PM UTC coverage: 63.12%. First build

Build # 18804694760

Build Type

Pull #7

github

Committed by

web-flow

Commit Message

Merge d3c293d65 into 43edde1f5

Pull Request Pull Request #7: Closes #5

Run Details

22 of 58 new or added lines in 9 files covered. (37.93%)

433 of 686 relevant lines covered (63.12%)

1.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/restcodegen/restclient/client.py

import json
import uuid
from json.decoder import JSONDecodeError
from typing import Any

import httpx
import structlog
from curlify2 import Curlify

from restcodegen.restclient.configuration import Configuration


class Logging:
    def __init__(self, configuration: Configuration) -> None:
        self.configuration = configuration
        self.log = structlog.get_logger(__name__).bind(service="api")

    def log_request(self, method: str, url: str, **kwargs: Any) -> None:
        log = self.log.bind(event_id=str(uuid.uuid4()))
        json_data = kwargs.get("json")
        content = kwargs.get("content")
        try:
            if content:
                json_data = json.loads(content)
        except JSONDecodeError:
            ...

        msg = dict(
            event="Request",
            method=method,
            path=url,
            host=self.configuration.base_url,
            params=kwargs.get("params"),
            headers=kwargs.get("headers"),
            data=kwargs.get("data"),
        )

        if isinstance(json_data, dict):
            msg["json"] = json_data

        log.msg(**msg)

    def log_response(self, response: httpx.Response) -> None:
        log = self.log.bind(event_id=str(uuid.uuid4()))
        curl = Curlify(response.request).to_curl()
        print(curl)
        log.msg(
            event="Response",
            status_code=response.status_code,
            headers=dict(response.headers),
            content=self._get_json(response),
        )

    @staticmethod
    def _get_json(response: httpx.Response) -> dict[str, Any] | bytes:
        try:
            return response.json()
        except JSONDecodeError:
            return response.content


class Client(httpx.Client):
    def __init__(self, configuration: Configuration) -> None:
        self.configuration = configuration
        super().__init__(base_url=self.configuration.base_url, **self.configuration.kwargs)
        self._logger = Logging(self.configuration)

    def request(self, method: str, url: str, **kwargs: Any) -> httpx.Response:  # type: ignore
        if not self.configuration.disable_log:
            self._logger.log_request(method, url, **kwargs)

        response = super().request(method, url, **kwargs)

        if not self.configuration.disable_log:
            self._logger.log_response(response)

        response.raise_for_status()
        return response


class AsyncClient(httpx.AsyncClient):
    def __init__(self, configuration: Configuration) -> None:
        self.configuration = configuration
        super().__init__(base_url=self.configuration.base_url, **self.configuration.kwargs)
        self._logger = Logging(configuration)

    async def request(self, method: str, url: str, **kwargs: Any) -> httpx.Response:  # type: ignore
        if not self.configuration.disable_log:
            self._logger.log_request(method, url, **kwargs)

        response = await super().request(method, url, **kwargs)

        if not self.configuration.disable_log:
            self._logger.log_response(response)

        response.raise_for_status()
        return response

1	import json	×
2	import uuid	×
3	from json.decoder import JSONDecodeError	×
4	from typing import Any	×
5
6	import httpx	×
7	import structlog	×
8	from curlify2 import Curlify	×
9
10	from restcodegen.restclient.configuration import Configuration	×
11
12
13	class Logging:	×
14	def __init__(self, configuration: Configuration) -> None:	×
15	self.configuration = configuration	×
16	self.log = structlog.get_logger(__name__).bind(service="api")	×
17
18	def log_request(self, method: str, url: str, **kwargs: Any) -> None:	×
19	log = self.log.bind(event_id=str(uuid.uuid4()))	×
20	json_data = kwargs.get("json")	×
21	content = kwargs.get("content")	×
22	try:	×
23	if content:	×
24	json_data = json.loads(content)	×
25	except JSONDecodeError:	×
26	...	×
27
28	msg = dict(	×
29	event="Request",
30	method=method,
31	path=url,
32	host=self.configuration.base_url,
33	params=kwargs.get("params"),
34	headers=kwargs.get("headers"),
35	data=kwargs.get("data"),
36	)
37
38	if isinstance(json_data, dict):	×
39	msg["json"] = json_data	×
40
41	log.msg(**msg)	×
42
43	def log_response(self, response: httpx.Response) -> None:	×
44	log = self.log.bind(event_id=str(uuid.uuid4()))	×
45	curl = Curlify(response.request).to_curl()	×
46	print(curl)	×
47	log.msg(	×
48	event="Response",
49	status_code=response.status_code,
50	headers=dict(response.headers),
51	content=self._get_json(response),
52	)
53
54	@staticmethod	×
55	def _get_json(response: httpx.Response) -> dict[str, Any] \| bytes:	×
56	try:	×
57	return response.json()	×
58	except JSONDecodeError:	×
59	return response.content	×
60
61
62	class Client(httpx.Client):	×
63	def __init__(self, configuration: Configuration) -> None:	×
64	self.configuration = configuration	×
NEW 65	super().__init__(base_url=self.configuration.base_url, **self.configuration.kwargs)	×
66	self._logger = Logging(self.configuration)	×
67
68	def request(self, method: str, url: str, **kwargs: Any) -> httpx.Response: # type: ignore	×
69	if not self.configuration.disable_log:	×
70	self._logger.log_request(method, url, **kwargs)	×
71
72	response = super().request(method, url, **kwargs)	×
73
74	if not self.configuration.disable_log:	×
75	self._logger.log_response(response)	×
76
77	response.raise_for_status()	×
78	return response	×
79
80
81	class AsyncClient(httpx.AsyncClient):	×
82	def __init__(self, configuration: Configuration) -> None:	×
83	self.configuration = configuration	×
NEW 84	super().__init__(base_url=self.configuration.base_url, **self.configuration.kwargs)	×
85	self._logger = Logging(configuration)	×
86
87	async def request(self, method: str, url: str, **kwargs: Any) -> httpx.Response: # type: ignore	×
88	if not self.configuration.disable_log:	×
89	self._logger.log_request(method, url, **kwargs)	×
90
91	response = await super().request(method, url, **kwargs)	×
92
93	if not self.configuration.disable_log:	×
94	self._logger.log_response(response)	×
95
96	response.raise_for_status()	×
97	return response	×

ValeriyMenshikov / restcodegen / 18804694760

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous