15833858062

Committed 23 Jun 2025 07:59PM UTC coverage: 64.627%. First build

Build # 15833858062

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1 from ValeriyMenshikov/manual_refactor

Manual refactor

Run Details

260 of 293 new or added lines in 12 files covered. (88.74%)

433 of 670 relevant lines covered (64.63%)

0.65 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	×
NEW 2	import uuid	×
3	from json.decoder import JSONDecodeError	×
NEW 4	from typing import Any	×
5
NEW 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	×
65	super().__init__(	×
66	base_url=self.configuration.base_url, **self.configuration.kwargs
67	)
68	self._logger = Logging(self.configuration)	×
69
70	def request(self, method: str, url: str, **kwargs: Any) -> httpx.Response: # type: ignore	×
71	if not self.configuration.disable_log:	×
72	self._logger.log_request(method, url, **kwargs)	×
73
74	response = super().request(method, url, **kwargs)	×
75
76	if not self.configuration.disable_log:	×
77	self._logger.log_response(response)	×
78
79	response.raise_for_status()	×
80	return response	×
81
82
83	class AsyncClient(httpx.AsyncClient):	×
84	def __init__(self, configuration: Configuration) -> None:	×
85	self.configuration = configuration	×
86	super().__init__(	×
87	base_url=self.configuration.base_url, **self.configuration.kwargs
88	)
89	self._logger = Logging(configuration)	×
90
91	async def request(self, method: str, url: str, **kwargs: Any) -> httpx.Response: # type: ignore	×
92	if not self.configuration.disable_log:	×
93	self._logger.log_request(method, url, **kwargs)	×
94
95	response = await super().request(method, url, **kwargs)	×
96
97	if not self.configuration.disable_log:	×
98	self._logger.log_response(response)	×
99
100	response.raise_for_status()	×
101	return response	×

ValeriyMenshikov / restcodegen / 15833858062

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