Skip to content

Urllib Client

urllib_client

Standard-library HTTP client implementation of HttpClientPort.

Uses http.client wrapped in asyncio.to_thread() to provide an
async HTTP client with zero external dependencies.

This adapter is constrained to str request/response bodies — it natively
encodes request bodies as UTF-8 and decodes response bodies as UTF-8. For
other content types (bytes, JSON, etc.), use a dedicated adapter or a
wrapper that handles serialization.

URLLibClient

Bases: HttpClientPort[str, str]

HTTP client backed by Python's http.client + asyncio.to_thread.

This adapter provides async HTTP methods without requiring external
dependencies like httpx or aiohttp. Request and response bodies
are encoded/decoded as UTF-8 strings.

Note

For production use with high concurrency or non-string payloads,
consider an adapter backed by httpx or aiohttp instead.

The module and class names reference urllib for historical
compatibility; the implementation uses http.client internally.

Raises:

Type Description
OSError

On network or connection failures.
HTTPException

On HTTP protocol errors.
ConfigurationError

On misconfigured URLs (e.g., non-HTTP schemes).
Example 
client = URLLibClient()
html = await client.get("https://example.com")
response = await client.post("https://httpbin.org/post", body="hello")
Source code in src/forging_blocks/infrastructure/http_client/urllib_client.py 
 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
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class URLLibClient(HttpClientPort[str, str]):
    """HTTP client backed by Python's ``http.client`` + ``asyncio.to_thread``.

    This adapter provides async HTTP methods without requiring external
    dependencies like ``httpx`` or ``aiohttp``. Request and response bodies
    are encoded/decoded as UTF-8 strings.

    Note:
        For production use with high concurrency or non-string payloads,
        consider an adapter backed by ``httpx`` or ``aiohttp`` instead.

        The module and class names reference ``urllib`` for historical
        compatibility; the implementation uses ``http.client`` internally.

    Raises:
        OSError: On network or connection failures.
        http.client.HTTPException: On HTTP protocol errors.
        ConfigurationError: On misconfigured URLs (e.g., non-HTTP schemes).

    Example:
        ```python
        client = URLLibClient()
        html = await client.get("https://example.com")
        response = await client.post("https://httpbin.org/post", body="hello")
        ```
    """

    async def request(
        self,
        method: str,
        url: str,
        headers: dict[str, str] | None = None,
        body: str | None = None,
    ) -> str:
        """Send an HTTP request and return the response body as a string.

        Args:
            method: HTTP method (GET, POST, PUT, DELETE, etc.).
            url: The target URL.
            headers: Optional HTTP headers.
            body: Optional request body string (UTF-8 encoded).

        Returns:
            The response body decoded as UTF-8 string.

        Raises:
            OSError: On network or connection failures.
            http.client.HTTPException: On HTTP protocol errors.

        """
        http_headers: dict[str, str] = headers or {}
        data: bytes | None = body.encode("utf-8") if body is not None else None
        parsed = urlparse(url)
        scheme = parsed.scheme
        if scheme == "http":
            Conn = HTTPConnection
        elif scheme == "https":
            Conn = HTTPSConnection
        else:
            raise ConfigurationError(
                f"Disallowed URL scheme '{scheme}'. Only http and https are supported."
            )

        def _do_request() -> str:
            conn = Conn(parsed.netloc)
            path = parsed.path or "/"
            if parsed.query:
                path = f"{path}?{parsed.query}"
            conn.request(method, path, body=data, headers=http_headers)
            with conn.getresponse() as response:
                return response.read().decode("utf-8")

        return await asyncio.to_thread(_do_request)

    async def get(
        self,
        url: str,
        headers: dict[str, str] | None = None,
    ) -> str:
        """Send an HTTP GET request."""
        return await self.request("GET", url, headers=headers)

    async def post(
        self,
        url: str,
        body: str | None = None,
        headers: dict[str, str] | None = None,
    ) -> str:
        """Send an HTTP POST request."""
        return await self.request("POST", url, headers=headers, body=body)

    async def put(
        self,
        url: str,
        body: str | None = None,
        headers: dict[str, str] | None = None,
    ) -> str:
        """Send an HTTP PUT request."""
        return await self.request("PUT", url, headers=headers, body=body)

    async def delete(
        self,
        url: str,
        headers: dict[str, str] | None = None,
    ) -> str:
        """Send an HTTP DELETE request."""
        return await self.request("DELETE", url, headers=headers)

request(method: str, url: str, headers: dict[str, str] | None = None, body: str | None = None) -> str async

Send an HTTP request and return the response body as a string.

Parameters:

Name Type Description Default
method str

HTTP method (GET, POST, PUT, DELETE, etc.).

 required
url str

The target URL.

 required
headers dict[str, str] | None

Optional HTTP headers.

 None
body str | None

Optional request body string (UTF-8 encoded).

 None

Returns:

Type Description
str

The response body decoded as UTF-8 string.

Raises:

Type Description
OSError

On network or connection failures.
HTTPException

On HTTP protocol errors.
Source code in src/forging_blocks/infrastructure/http_client/urllib_client.py 
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
93
94
async def request(
    self,
    method: str,
    url: str,
    headers: dict[str, str] | None = None,
    body: str | None = None,
) -> str:
    """Send an HTTP request and return the response body as a string.

    Args:
        method: HTTP method (GET, POST, PUT, DELETE, etc.).
        url: The target URL.
        headers: Optional HTTP headers.
        body: Optional request body string (UTF-8 encoded).

    Returns:
        The response body decoded as UTF-8 string.

    Raises:
        OSError: On network or connection failures.
        http.client.HTTPException: On HTTP protocol errors.

    """
    http_headers: dict[str, str] = headers or {}
    data: bytes | None = body.encode("utf-8") if body is not None else None
    parsed = urlparse(url)
    scheme = parsed.scheme
    if scheme == "http":
        Conn = HTTPConnection
    elif scheme == "https":
        Conn = HTTPSConnection
    else:
        raise ConfigurationError(
            f"Disallowed URL scheme '{scheme}'. Only http and https are supported."
        )

    def _do_request() -> str:
        conn = Conn(parsed.netloc)
        path = parsed.path or "/"
        if parsed.query:
            path = f"{path}?{parsed.query}"
        conn.request(method, path, body=data, headers=http_headers)
        with conn.getresponse() as response:
            return response.read().decode("utf-8")

    return await asyncio.to_thread(_do_request)

get(url: str, headers: dict[str, str] | None = None) -> str async

Send an HTTP GET request.

Source code in src/forging_blocks/infrastructure/http_client/urllib_client.py 
 96
 97
 98
 99
100
101
102
async def get(
    self,
    url: str,
    headers: dict[str, str] | None = None,
) -> str:
    """Send an HTTP GET request."""
    return await self.request("GET", url, headers=headers)

post(url: str, body: str | None = None, headers: dict[str, str] | None = None) -> str async

Send an HTTP POST request.

Source code in src/forging_blocks/infrastructure/http_client/urllib_client.py 
104
105
106
107
108
109
110
111
async def post(
    self,
    url: str,
    body: str | None = None,
    headers: dict[str, str] | None = None,
) -> str:
    """Send an HTTP POST request."""
    return await self.request("POST", url, headers=headers, body=body)

put(url: str, body: str | None = None, headers: dict[str, str] | None = None) -> str async

Send an HTTP PUT request.

Source code in src/forging_blocks/infrastructure/http_client/urllib_client.py 
113
114
115
116
117
118
119
120
async def put(
    self,
    url: str,
    body: str | None = None,
    headers: dict[str, str] | None = None,
) -> str:
    """Send an HTTP PUT request."""
    return await self.request("PUT", url, headers=headers, body=body)

delete(url: str, headers: dict[str, str] | None = None) -> str async

Send an HTTP DELETE request.

Source code in src/forging_blocks/infrastructure/http_client/urllib_client.py 
122
123
124
125
126
127
128
async def delete(
    self,
    url: str,
    headers: dict[str, str] | None = None,
) -> str:
    """Send an HTTP DELETE request."""
    return await self.request("DELETE", url, headers=headers)