Skip to content

# transports

## `xsdata.formats.dataclass.transports`

### `Transport`

Bases: `ABC`

An HTTP transport interface.

Source code in `xsdata/formats/dataclass/transports.py`

     7
     8
     9
    10
    11
    12
    13
    14
    15
    16
    17
    18

    class Transport(abc.ABC):
        """An HTTP transport interface."""

        __slots__ = ()

        @abc.abstractmethod
        def get(self, url: str, params: dict, headers: dict) -> bytes:
            """Send a GET request."""

        @abc.abstractmethod
        def post(self, url: str, data: Any, headers: dict) -> bytes:
            """Send a POST request."""

#### `get(url, params, headers)` `abstractmethod`

Send a GET request.

Source code in `xsdata/formats/dataclass/transports.py`

    12
    13
    14

    @abc.abstractmethod
    def get(self, url: str, params: dict, headers: dict) -> bytes:
        """Send a GET request."""

#### `post(url, data, headers)` `abstractmethod`

Send a POST request.

Source code in `xsdata/formats/dataclass/transports.py`

    16
    17
    18

    @abc.abstractmethod
    def post(self, url: str, data: Any, headers: dict) -> bytes:
        """Send a POST request."""

### `DefaultTransport`

Bases: `Transport`

Default transport based on the `requests` library.

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`timeout` | `float` | Read timeout in seconds | `2.0`

Source code in `xsdata/formats/dataclass/transports.py`

    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    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 DefaultTransport(Transport):
        """Default transport based on the `requests` library.

        Args:
            timeout: Read timeout in seconds
        """

        __slots__ = "session", "timeout"

        def __init__(self, timeout: float = 2.0, session: Session | None = None):
            """Initialize the transport."""
            self.timeout = timeout
            self.session = session or Session()

        def get(self, url: str, params: dict, headers: dict) -> bytes:
            """Send a GET request.

            Args:
                url: The base URL
                params: The query parameters
                headers: A key-value map of HTTP headers

            Returns:
                The encoded response content.

            Raises:
                HTTPError: if status code is not valid for content unmarshalling.
            """
            res = self.session.get(
                url,
                params=params,
                headers=headers,
                timeout=self.timeout,
            )
            return self.handle_response(res)

        def post(self, url: str, data: Any, headers: dict) -> Any:
            """Send a POST request.

            Args:
                url: The base URL
                data: The request body payload
                headers: A key-value map of HTTP headers

            Returns:
                The encoded response content.

            Raises:
                HTTPError: if status code is not valid for content unmarshalling.
            """
            res = self.session.post(url, data=data, headers=headers, timeout=self.timeout)
            return self.handle_response(res)

        @classmethod
        def handle_response(cls, response: Response) -> bytes:
            """Return the response content or raise an exception.

            Status codes 200 or 500 means that we can unmarshall the response.

            Args:
                response: The response instance

            Returns:
                The encoded response content.

            Raises:
                HTTPError: If the response status code is not 200 or 500
            """
            if response.status_code not in (200, 500):
                response.raise_for_status()

            return response.content

#### `__init__(timeout=2.0, session=None)`

Initialize the transport.

Source code in `xsdata/formats/dataclass/transports.py`

    30
    31
    32
    33

    def __init__(self, timeout: float = 2.0, session: Session | None = None):
        """Initialize the transport."""
        self.timeout = timeout
        self.session = session or Session()

#### `get(url, params, headers)`

Send a GET request.

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`url` | `str` | The base URL | required
`params` | `dict` | The query parameters | required
`headers` | `dict` | A key-value map of HTTP headers | required

Returns:

Type | Description
--- | ---
`bytes` | The encoded response content.

Raises:

Type | Description
--- | ---
`HTTPError` | if status code is not valid for content unmarshalling.

Source code in `xsdata/formats/dataclass/transports.py`

    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55

    def get(self, url: str, params: dict, headers: dict) -> bytes:
        """Send a GET request.

        Args:
            url: The base URL
            params: The query parameters
            headers: A key-value map of HTTP headers

        Returns:
            The encoded response content.

        Raises:
            HTTPError: if status code is not valid for content unmarshalling.
        """
        res = self.session.get(
            url,
            params=params,
            headers=headers,
            timeout=self.timeout,
        )
        return self.handle_response(res)

#### `post(url, data, headers)`

Send a POST request.

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`url` | `str` | The base URL | required
`data` | `Any` | The request body payload | required
`headers` | `dict` | A key-value map of HTTP headers | required

Returns:

Type | Description
--- | ---
`Any` | The encoded response content.

Raises:

Type | Description
--- | ---
`HTTPError` | if status code is not valid for content unmarshalling.

Source code in `xsdata/formats/dataclass/transports.py`

    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72

    def post(self, url: str, data: Any, headers: dict) -> Any:
        """Send a POST request.

        Args:
            url: The base URL
            data: The request body payload
            headers: A key-value map of HTTP headers

        Returns:
            The encoded response content.

        Raises:
            HTTPError: if status code is not valid for content unmarshalling.
        """
        res = self.session.post(url, data=data, headers=headers, timeout=self.timeout)
        return self.handle_response(res)

#### `handle_response(response)` `classmethod`

Return the response content or raise an exception.

Status codes 200 or 500 means that we can unmarshall the response.

Parameters:

Name | Type | Description | Default
--- | --- | --- | ---
`response` | `Response` | The response instance | required

Returns:

Type | Description
--- | ---
`bytes` | The encoded response content.

Raises:

Type | Description
--- | ---
`HTTPError` | If the response status code is not 200 or 500

Source code in `xsdata/formats/dataclass/transports.py`

    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92

    @classmethod
    def handle_response(cls, response: Response) -> bytes:
        """Return the response content or raise an exception.

        Status codes 200 or 500 means that we can unmarshall the response.

        Args:
            response: The response instance

        Returns:
            The encoded response content.

        Raises:
            HTTPError: If the response status code is not 200 or 500
        """
        if response.status_code not in (200, 500):
            response.raise_for_status()

        return response.content