Skip to content

Transport

Transport

Bases: ABC

Protocol / base class for HTTP transports.

Implementations must be reusable across multiple requests.

Source code in merchants/transport.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
class Transport(ABC):
    """Protocol / base class for HTTP transports.

    Implementations must be reusable across multiple requests.
    """

    @abstractmethod
    def send(
        self,
        method: str,
        url: str,
        *,
        headers: dict[str, str] | None = None,
        json: Any = None,
        params: dict[str, str] | None = None,
        timeout: float = 30.0,
    ) -> HttpResponse:
        """Send an HTTP request and return an :class:`HttpResponse`.

        Raises:
            TransportError: On network or connection failure.
        """

Functions

send abstractmethod 

send(
    method: str,
    url: str,
    *,
    headers: dict[str, str] | None = None,
    json: Any = None,
    params: dict[str, str] | None = None,
    timeout: float = 30.0,
) -> HttpResponse

Send an HTTP request and return an :class:HttpResponse.

Raises:

Type Description
TransportError

On network or connection failure.
Source code in merchants/transport.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
@abstractmethod
def send(
    self,
    method: str,
    url: str,
    *,
    headers: dict[str, str] | None = None,
    json: Any = None,
    params: dict[str, str] | None = None,
    timeout: float = 30.0,
) -> HttpResponse:
    """Send an HTTP request and return an :class:`HttpResponse`.

    Raises:
        TransportError: On network or connection failure.
    """

HttpResponse

Thin wrapper around an HTTP response.

Source code in merchants/transport.py 
15
16
17
18
19
20
21
22
23
24
25
class HttpResponse:
    """Thin wrapper around an HTTP response."""

    def __init__(self, status_code: int, headers: dict[str, str], body: Any) -> None:
        self.status_code = status_code
        self.headers = headers
        self.body = body

    @property
    def ok(self) -> bool:
        return 200 <= self.status_code < 300

RequestsTransport

Bases: Transport

Default transport backed by :mod:requests.

A single :class:requests.Session is reused for connection pooling.

Source code in merchants/transport.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
class RequestsTransport(Transport):
    """Default transport backed by :mod:`requests`.

    A single :class:`requests.Session` is reused for connection pooling.
    """

    def __init__(self, session: requests.Session | None = None) -> None:
        self._session = session or requests.Session()

    def send(
        self,
        method: str,
        url: str,
        *,
        headers: dict[str, str] | None = None,
        json: Any = None,
        params: dict[str, str] | None = None,
        timeout: float = 30.0,
    ) -> HttpResponse:
        try:
            resp = self._session.request(
                method,
                url,
                headers=headers,
                json=json,
                params=params,
                timeout=timeout,
            )
        except requests.RequestException as exc:
            raise TransportError(str(exc)) from exc

        try:
            body = resp.json()
        except ValueError:
            body = resp.text

        return HttpResponse(
            status_code=resp.status_code,
            headers=dict(resp.headers),
            body=body,
        )

TransportError

Bases: Exception

Raised when an HTTP-level or network-level error occurs.

Source code in merchants/transport.py 
11
12
class TransportError(Exception):
    """Raised when an HTTP-level or network-level error occurs."""