Skip to content

SSF Tools - HTTP Client Service Architecture

Overview

The HTTP Client Service provides a unified, protocol-based approach to HTTP operations across all SSF Tools commands. This architecture enables consistent network handling while maintaining loose coupling through dependency injection patterns and supporting various HTTP operations including requests, file downloads, retry logic, caching, and error recovery with beautiful terminal feedback.

Architectural Principles

1. Protocol-Based Design

All HTTP client services implement well-defined protocols, enabling:

  • Dependency injection: Services receive HTTP client interfaces, not concrete implementations
  • Testability: Easy mocking and stubbing for unit tests
  • Flexibility: Swappable implementations (different HTTP libraries, mock clients)
  • Type safety: MyPy-compliant interfaces with clear contracts
  • Composability: Small, focused protocols that can be combined

2. Separation of Concerns

Each HTTP operation is handled through a dedicated interface:

  • Request Handling: GET, POST, PUT, DELETE, HEAD, PATCH operations
  • File Downloads: Progress tracking and streaming downloads
  • Retry Logic: Configurable retry strategies with backoff algorithms
  • Caching: Response caching with TTL-based expiration
  • Error Recovery: Graceful degradation and comprehensive error handling
  • Progress Feedback: Rich terminal output integration for user experience

3. Robust Error Handling

All HTTP services implement graceful error handling:

  • Network errors: Connection timeouts, DNS failures, SSL issues
  • HTTP errors: 4xx client errors, 5xx server errors with context
  • Retry strategies: Automatic retry with exponential backoff
  • Type safety: Clear exception hierarchy for different error types

Architecture Overview

The HTTP Client Service follows a layered architecture with clear separation of concerns:

Service Composition

graph TD Client[Client Code] --> HCS[HttpClientService] HCS --> HC[HttpConfig] HCS --> RO[RichOutputService] HCS --> HTTPX[HTTPX Client Library] HCS --> Cache[Response Cache] HCS --> RS[Retry Strategy] HCS --> EH[Error Handler] HCS --> PH[Progress Handler] classDef service fill:#e1f5fe,stroke:#0277bd,stroke-width:2px classDef config fill:#f3e5f5,stroke:#7b1fa2,stroke-width:2px classDef external fill:#fce4ec,stroke:#c2185b,stroke-width:2px classDef handler fill:#e8f5e8,stroke:#2e7d32,stroke-width:2px class HCS service class HC config class RO,HTTPX,Cache external class RS,EH,PH handler

Core Protocols

graph LR subgraph "HTTP Operations" HCP["HttpClientProtocol<br/>- get()<br/>- post()<br/>- download_file()"] UP["UrlProtocol<br/>- is_url_accessible()"] SP["SessionProtocol<br/>- session context manager"] end subgraph "Configuration" HC["HttpConfig<br/>- timeout_seconds<br/>- max_retries<br/>- retry_strategy"] HR["HttpResponse<br/>- status_code<br/>- content<br/>- elapsed_seconds"] end subgraph "Error Handling" NE["NetworkError<br/>- base exception"] TE["TimeoutError<br/>- request timeouts"] CE["ConnectionError<br/>- connection issues"] HE["HttpError<br/>- HTTP protocol errors"] end classDef protocol fill:#f3e5f5,stroke:#7b1fa2,stroke-width:2px class HCP,UP,SP,HC,HR,NE,TE,CE,HE protocol

Implementation Architecture

graph TD subgraph "HTTP Client Implementation" HCS[HttpClientService] -.-> HCSI[HttpClientService Implementation] end subgraph "Configuration Models" HC[HttpConfig] -.-> HCI[HttpConfig Implementation] HR[HttpResponse] -.-> HRI[HttpResponse Implementation] end subgraph "Retry Strategies" RS[RetryStrategy] -.-> EB[ExponentialBackoff] RS -.-> LB[LinearBackoff] RS -.-> FD[FixedDelay] end classDef protocol fill:#f3e5f5,stroke:#7b1fa2,stroke-width:2px classDef implementation fill:#e8f5e8,stroke:#2e7d32,stroke-width:2px class HCS,HC,HR,RS protocol class HCSI,HCI,HRI,EB,LB,FD implementation

Dependency Injection

graph TB CC[Core Container] --> HCS[HttpClientService] CC --> HC[HttpConfig Factory] CC --> RO[RichOutputService] HC --> HCS RO --> HCS HCS --> FPS[FileProcessingService] classDef container fill:#fff3e0,stroke:#ef6c00,stroke-width:2px classDef service fill:#e1f5fe,stroke:#0277bd,stroke-width:2px classDef config fill:#f3e5f5,stroke:#7b1fa2,stroke-width:2px classDef dependent fill:#e8f5e8,stroke:#2e7d32,stroke-width:2px class CC container class HCS service class HC,RO config class FPS dependent

HTTP Client Workflow

The HTTP client service orchestrates network operations with comprehensive error handling and progress feedback:

sequenceDiagram participant Client participant HCS as HttpClientService participant RO as RichOutputService participant HTTPX as HTTPX Client participant Cache as Response Cache Client->>HCS: get(url, headers) HCS->>Cache: check_cache(url) Cache-->>HCS: cache_result opt Cache Miss HCS->>RO: info("Making HTTP request...") HCS->>HTTPX: request(method, url, headers) alt Success HTTPX-->>HCS: response HCS->>Cache: store(url, response) HCS->>RO: success("Request completed") else Network Error HCS->>RO: warning("Retrying request...") HCS->>HTTPX: retry_request() alt Retry Success HTTPX-->>HCS: response HCS->>RO: success("Request succeeded after retry") else Retry Failed HCS->>RO: error("Request failed after retries") HCS-->>Client: NetworkError end end end HCS-->>Client: HttpResponse

Core Components

kp_ssf_tools.core.services.http_client.service.HttpClientService

HTTP client service implementing HttpClientProtocol.

Source code in src\kp_ssf_tools\core\services\http_client\service.py 
 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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
class HttpClientService:
    """HTTP client service implementing HttpClientProtocol."""

    def __init__(
        self,
        config: HttpConfig | None = None,
        output: RichOutputProtocol | None = None,
    ) -> None:
        """
        Initialize HTTP client service.

        Args:
            config: HTTP client configuration
            output: Rich output service for logging and user feedback

        """
        self.config = config or HttpConfig()
        self.output = output
        self._client: httpx.Client | None = None
        self._cache: dict[str, tuple[HttpResponse, float]] = {}

    def _create_client(self) -> httpx.Client:
        """Create configured httpx client."""
        return httpx.Client(
            timeout=self.config.timeout_seconds,
            follow_redirects=self.config.follow_redirects,
            verify=self.config.verify_ssl,
            limits=httpx.Limits(max_connections=self.config.max_connections),
            headers={"User-Agent": self.config.user_agent},
        )

    @property
    def client(self) -> httpx.Client:
        """Get or create HTTP client."""
        if self._client is None:
            self._client = self._create_client()
        return self._client

    def get(
        self,
        url: str,
        headers: dict[str, str] | None = None,
        params: dict[str, str] | None = None,
        **kwargs: object,
    ) -> HttpResponse:
        """
        Execute GET request.

        Args:
            url: Target URL
            headers: Optional request headers
            params: Optional query parameters
            **kwargs: Additional request options

        Returns:
            HTTP response wrapper

        Raises:
            NetworkError: When request fails

        """
        return self._execute_with_retry(
            RequestMethod.GET,
            url,
            headers=headers,
            params=params,
            **kwargs,
        )

    def post(
        self,
        url: str,
        data: bytes | str | dict[str, object] | None = None,
        json: dict[str, object] | None = None,
        headers: dict[str, str] | None = None,
        **kwargs: object,
    ) -> HttpResponse:
        """
        Execute POST request.

        Args:
            url: Target URL
            data: Request body data
            json: JSON data to send
            headers: Optional request headers
            **kwargs: Additional request options

        Returns:
            HTTP response wrapper

        Raises:
            NetworkError: When request fails

        """
        return self._execute_with_retry(
            RequestMethod.POST,
            url,
            data=data,
            json=json,
            headers=headers,
            **kwargs,
        )

    def head(
        self,
        url: str,
        headers: dict[str, str] | None = None,
        **kwargs: object,
    ) -> HttpResponse:
        """
        Execute HEAD request.

        Args:
            url: Target URL
            headers: Optional request headers
            **kwargs: Additional request options

        Returns:
            HTTP response wrapper

        Raises:
            NetworkError: When request fails

        """
        return self._execute_with_retry(
            RequestMethod.HEAD,
            url,
            headers=headers,
            **kwargs,
        )

    def download_file(
        self,
        url: str,
        file_path: Path,
        progress_callback: Callable[[int, int], None] | None = None,
    ) -> HttpResponse:
        """
        Download file with optional progress tracking.

        Args:
            url: File URL to download
            file_path: Destination file path
            progress_callback: Optional progress callback function

        Returns:
            HTTP response wrapper

        Raises:
            NetworkError: When download fails

        """

        def download_attempt() -> HttpResponse:
            """Single download attempt."""
            start_time = time.time()

            try:
                with self.client.stream("GET", url) as response:
                    response.raise_for_status()

                    # Get content length for progress tracking
                    content_length = int(response.headers.get("content-length", 0))

                    # Ensure parent directory exists
                    file_path.parent.mkdir(parents=True, exist_ok=True)

                    downloaded = 0
                    with file_path.open("wb") as file:
                        for chunk in response.iter_bytes(chunk_size=8192):
                            file.write(chunk)
                            downloaded += len(chunk)

                            if progress_callback and content_length > 0:
                                progress_callback(downloaded, content_length)

                    elapsed = time.time() - start_time

                    return HttpResponse(
                        status_code=response.status_code,
                        headers=dict(response.headers),
                        content=b"",  # Don't store large file content
                        text="",
                        url=str(response.url),
                        elapsed_seconds=elapsed,
                    )

            except httpx.TimeoutException as exc:
                msg = f"Download timeout for {url}"
                raise HttpTimeoutError(msg, url=url) from exc
            except httpx.ConnectError as exc:
                msg = f"Connection failed for {url}"
                raise HttpConnectionError(
                    msg,
                    url=url,
                ) from exc
            except httpx.HTTPStatusError as exc:
                msg = f"HTTP error {exc.response.status_code} for {url}"
                raise HttpError(
                    msg,
                    url=url,
                    status_code=exc.response.status_code,
                ) from exc
            except Exception as exc:
                msg = f"Download failed for {url}"
                raise NetworkError(msg, url=url) from exc

        return self._retry_operation(download_attempt, url)

    def is_url_accessible(self, url: str) -> bool:
        """
        Check if URL is accessible without downloading content.

        Args:
            url: URL to check

        Returns:
            True if URL is accessible, False otherwise

        """
        try:
            response = self.head(url)
        except NetworkError:
            return False
        else:
            return response.is_success

    @contextmanager
    def session(self) -> Generator[httpx.Client]:
        """
        Context manager for persistent session.

        Yields:
            HTTP client session

        """
        client = self._create_client()
        try:
            yield client
        finally:
            client.close()

    def clear_cache(self) -> None:
        """Clear the response cache."""
        self._cache.clear()

    def _execute_with_retry(
        self,
        method: RequestMethod,
        url: str,
        **kwargs: object,
    ) -> HttpResponse:
        """Execute HTTP request with retry logic."""

        def request_attempt() -> HttpResponse:
            """Single request attempt."""
            start_time = time.time()

            try:
                response = self.client.request(method.value, url, **kwargs)  # type: ignore[arg-type]
                elapsed = time.time() - start_time

                return HttpResponse(
                    status_code=response.status_code,
                    headers=dict(response.headers),
                    content=response.content,
                    text=response.text,
                    url=str(response.url),
                    elapsed_seconds=elapsed,
                )

            except httpx.TimeoutException as exc:
                msg = f"Request timeout for {url}"
                raise HttpTimeoutError(msg, url=url) from exc
            except httpx.ConnectError as exc:
                msg = f"Connection failed for {url}"
                raise HttpConnectionError(msg, url=url) from exc
            except httpx.HTTPStatusError as exc:
                msg = f"HTTP error {exc.response.status_code} for {url}"
                raise HttpError(
                    msg,
                    url=url,
                    status_code=exc.response.status_code,
                ) from exc
            except Exception as exc:
                msg = f"Request failed for {url}"
                raise NetworkError(msg, url=url) from exc

        return self._retry_operation(request_attempt, url)

    def _retry_operation(
        self,
        operation: Callable[[], HttpResponse],
        url: str,
    ) -> HttpResponse:
        """Execute operation with configured retry strategy."""
        last_exception: NetworkError | None = None

        for attempt in range(self.config.max_retries + 1):
            try:
                response = operation()

                # Mark successful retry if we had previous failures
                if last_exception and attempt > 0 and self.output:
                    self.output.success(
                        f"Request succeeded after {attempt} retry attempts for {url}",
                    )

            except NetworkError as exc:
                last_exception = exc
                exc.retry_attempted = attempt > 0

                # Don't retry on final attempt
                if attempt == self.config.max_retries:
                    break

                # Calculate delay for next attempt
                delay = self._calculate_retry_delay(attempt + 1)

                # Log retry attempt with delay information
                if self.output:
                    self.output.warning(
                        f"Request failed for {url}, retrying in {delay:.1f}s "
                        f"(attempt {attempt + 1}/{self.config.max_retries})",
                    )

                time.sleep(delay)
            else:
                return response

        # All retries exhausted
        if last_exception:
            raise last_exception

        # Should not reach here, but just in case
        msg = f"Request failed for {url}"
        raise NetworkError(msg, url=url)

    def _calculate_retry_delay(self, attempt: int) -> float:
        """Calculate delay for retry attempt based on strategy."""
        if self.config.retry_strategy == RetryStrategy.NONE:
            return 0.0

        base_delay = self.config.base_delay_seconds

        if self.config.retry_strategy == RetryStrategy.EXPONENTIAL_BACKOFF:
            delay = base_delay * (2 ** (attempt - 1))
        elif self.config.retry_strategy == RetryStrategy.LINEAR_BACKOFF:
            delay = base_delay * attempt
        else:  # FIXED_DELAY
            delay = base_delay

        return float(min(delay, self.config.max_delay_seconds))

    def __enter__(self) -> "HttpClientService":
        """Context manager entry."""
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        """Context manager exit with cleanup."""
        if self._client:
            self._client.close()
            self._client = None

Attributes

client property

Get or create HTTP client.

Functions

__enter__()

Context manager entry.

Source code in src\kp_ssf_tools\core\services\http_client\service.py 
378
379
380
def __enter__(self) -> "HttpClientService":
    """Context manager entry."""
    return self
__exit__(exc_type, exc_val, exc_tb)

Context manager exit with cleanup.

Source code in src\kp_ssf_tools\core\services\http_client\service.py 
382
383
384
385
386
387
388
389
390
391
def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Context manager exit with cleanup."""
    if self._client:
        self._client.close()
        self._client = None
__init__(config=None, output=None)

Initialize HTTP client service.

Parameters:

Name Type Description Default
config HttpConfig | None

HTTP client configuration

 None
output RichOutputProtocol | None

Rich output service for logging and user feedback

 None
Source code in src\kp_ssf_tools\core\services\http_client\service.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def __init__(
    self,
    config: HttpConfig | None = None,
    output: RichOutputProtocol | None = None,
) -> None:
    """
    Initialize HTTP client service.

    Args:
        config: HTTP client configuration
        output: Rich output service for logging and user feedback

    """
    self.config = config or HttpConfig()
    self.output = output
    self._client: httpx.Client | None = None
    self._cache: dict[str, tuple[HttpResponse, float]] = {}
clear_cache()

Clear the response cache.

Source code in src\kp_ssf_tools\core\services\http_client\service.py 
266
267
268
def clear_cache(self) -> None:
    """Clear the response cache."""
    self._cache.clear()
download_file(url, file_path, progress_callback=None)

Download file with optional progress tracking.

Parameters:

Name Type Description Default
url str

File URL to download

 required
file_path Path

Destination file path

 required
progress_callback Callable[[int, int], None] | None

Optional progress callback function

 None

Returns:

Type Description
HttpResponse

HTTP response wrapper

Raises:

Type Description
NetworkError

When download fails
Source code in src\kp_ssf_tools\core\services\http_client\service.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def download_file(
    self,
    url: str,
    file_path: Path,
    progress_callback: Callable[[int, int], None] | None = None,
) -> HttpResponse:
    """
    Download file with optional progress tracking.

    Args:
        url: File URL to download
        file_path: Destination file path
        progress_callback: Optional progress callback function

    Returns:
        HTTP response wrapper

    Raises:
        NetworkError: When download fails

    """

    def download_attempt() -> HttpResponse:
        """Single download attempt."""
        start_time = time.time()

        try:
            with self.client.stream("GET", url) as response:
                response.raise_for_status()

                # Get content length for progress tracking
                content_length = int(response.headers.get("content-length", 0))

                # Ensure parent directory exists
                file_path.parent.mkdir(parents=True, exist_ok=True)

                downloaded = 0
                with file_path.open("wb") as file:
                    for chunk in response.iter_bytes(chunk_size=8192):
                        file.write(chunk)
                        downloaded += len(chunk)

                        if progress_callback and content_length > 0:
                            progress_callback(downloaded, content_length)

                elapsed = time.time() - start_time

                return HttpResponse(
                    status_code=response.status_code,
                    headers=dict(response.headers),
                    content=b"",  # Don't store large file content
                    text="",
                    url=str(response.url),
                    elapsed_seconds=elapsed,
                )

        except httpx.TimeoutException as exc:
            msg = f"Download timeout for {url}"
            raise HttpTimeoutError(msg, url=url) from exc
        except httpx.ConnectError as exc:
            msg = f"Connection failed for {url}"
            raise HttpConnectionError(
                msg,
                url=url,
            ) from exc
        except httpx.HTTPStatusError as exc:
            msg = f"HTTP error {exc.response.status_code} for {url}"
            raise HttpError(
                msg,
                url=url,
                status_code=exc.response.status_code,
            ) from exc
        except Exception as exc:
            msg = f"Download failed for {url}"
            raise NetworkError(msg, url=url) from exc

    return self._retry_operation(download_attempt, url)
get(url, headers=None, params=None, **kwargs)

Execute GET request.

Parameters:

Name Type Description Default
url str

Target URL

 required
headers dict[str, str] | None

Optional request headers

 None
params dict[str, str] | None

Optional query parameters

 None
**kwargs object

Additional request options

 {}

Returns:

Type Description
HttpResponse

HTTP response wrapper

Raises:

Type Description
NetworkError

When request fails
Source code in src\kp_ssf_tools\core\services\http_client\service.py 
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
def get(
    self,
    url: str,
    headers: dict[str, str] | None = None,
    params: dict[str, str] | None = None,
    **kwargs: object,
) -> HttpResponse:
    """
    Execute GET request.

    Args:
        url: Target URL
        headers: Optional request headers
        params: Optional query parameters
        **kwargs: Additional request options

    Returns:
        HTTP response wrapper

    Raises:
        NetworkError: When request fails

    """
    return self._execute_with_retry(
        RequestMethod.GET,
        url,
        headers=headers,
        params=params,
        **kwargs,
    )
head(url, headers=None, **kwargs)

Execute HEAD request.

Parameters:

Name Type Description Default
url str

Target URL

 required
headers dict[str, str] | None

Optional request headers

 None
**kwargs object

Additional request options

 {}

Returns:

Type Description
HttpResponse

HTTP response wrapper

Raises:

Type Description
NetworkError

When request fails
Source code in src\kp_ssf_tools\core\services\http_client\service.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def head(
    self,
    url: str,
    headers: dict[str, str] | None = None,
    **kwargs: object,
) -> HttpResponse:
    """
    Execute HEAD request.

    Args:
        url: Target URL
        headers: Optional request headers
        **kwargs: Additional request options

    Returns:
        HTTP response wrapper

    Raises:
        NetworkError: When request fails

    """
    return self._execute_with_retry(
        RequestMethod.HEAD,
        url,
        headers=headers,
        **kwargs,
    )
is_url_accessible(url)

Check if URL is accessible without downloading content.

Parameters:

Name Type Description Default
url str

URL to check

 required

Returns:

Type Description
bool

True if URL is accessible, False otherwise
Source code in src\kp_ssf_tools\core\services\http_client\service.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def is_url_accessible(self, url: str) -> bool:
    """
    Check if URL is accessible without downloading content.

    Args:
        url: URL to check

    Returns:
        True if URL is accessible, False otherwise

    """
    try:
        response = self.head(url)
    except NetworkError:
        return False
    else:
        return response.is_success
post(url, data=None, json=None, headers=None, **kwargs)

Execute POST request.

Parameters:

Name Type Description Default
url str

Target URL

 required
data bytes | str | dict[str, object] | None

Request body data

 None
json dict[str, object] | None

JSON data to send

 None
headers dict[str, str] | None

Optional request headers

 None
**kwargs object

Additional request options

 {}

Returns:

Type Description
HttpResponse

HTTP response wrapper

Raises:

Type Description
NetworkError

When request fails
Source code in src\kp_ssf_tools\core\services\http_client\service.py 
 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
def post(
    self,
    url: str,
    data: bytes | str | dict[str, object] | None = None,
    json: dict[str, object] | None = None,
    headers: dict[str, str] | None = None,
    **kwargs: object,
) -> HttpResponse:
    """
    Execute POST request.

    Args:
        url: Target URL
        data: Request body data
        json: JSON data to send
        headers: Optional request headers
        **kwargs: Additional request options

    Returns:
        HTTP response wrapper

    Raises:
        NetworkError: When request fails

    """
    return self._execute_with_retry(
        RequestMethod.POST,
        url,
        data=data,
        json=json,
        headers=headers,
        **kwargs,
    )
session()

Context manager for persistent session.

Yields:

Type Description
Generator[Client]

HTTP client session
Source code in src\kp_ssf_tools\core\services\http_client\service.py 
251
252
253
254
255
256
257
258
259
260
261
262
263
264
@contextmanager
def session(self) -> Generator[httpx.Client]:
    """
    Context manager for persistent session.

    Yields:
        HTTP client session

    """
    client = self._create_client()
    try:
        yield client
    finally:
        client.close()

HTTP Configuration

kp_ssf_tools.core.services.http_client.models.HttpConfig dataclass

HTTP client configuration.

Source code in src\kp_ssf_tools\core\services\http_client\models.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
@dataclass
class HttpConfig:
    """HTTP client configuration."""

    timeout_seconds: float = 10.0
    max_retries: int = 3
    retry_strategy: RetryStrategy = RetryStrategy.EXPONENTIAL_BACKOFF
    base_delay_seconds: float = 1.0
    max_delay_seconds: float = 60.0
    user_agent: str = "SSF-Tools/1.0"
    follow_redirects: bool = True
    verify_ssl: bool = True
    max_connections: int = 10
    cache_enabled: bool = True
    cache_ttl_seconds: int = 3600

kp_ssf_tools.core.services.http_client.models.HttpResponse dataclass

HTTP response wrapper.

Source code in src\kp_ssf_tools\core\services\http_client\models.py 
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
@dataclass
class HttpResponse:
    """HTTP response wrapper."""

    status_code: int
    headers: dict[str, str]
    content: bytes
    text: str
    url: str
    elapsed_seconds: float
    from_cache: bool = False

    @property
    def is_success(self) -> bool:
        """Check if response indicates success."""
        return HTTP_STATUS_OK <= self.status_code < HTTP_STATUS_REDIRECT_START

    @property
    def is_client_error(self) -> bool:
        """Check if response indicates client error."""
        return (
            HTTP_STATUS_CLIENT_ERROR_START
            <= self.status_code
            < HTTP_STATUS_SERVER_ERROR_START
        )

    @property
    def is_server_error(self) -> bool:
        """Check if response indicates server error."""
        return (
            HTTP_STATUS_SERVER_ERROR_START
            <= self.status_code
            < HTTP_STATUS_SERVER_ERROR_END
        )
Attributes
is_client_error property

Check if response indicates client error.

is_server_error property

Check if response indicates server error.

is_success property

Check if response indicates success.

HTTP Client Interface

kp_ssf_tools.core.services.http_client.interfaces.HttpClientProtocol

Bases: Protocol

Protocol defining the HTTP client interface for dependency injection.

Source code in src\kp_ssf_tools\core\services\http_client\interfaces.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 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
 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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
@runtime_checkable
class HttpClientProtocol(Protocol):
    """Protocol defining the HTTP client interface for dependency injection."""

    @abstractmethod
    def get(
        self,
        url: str,
        headers: dict[str, str] | None = None,
        params: dict[str, str] | None = None,
        **kwargs: object,
    ) -> HttpResponse:
        """
        Execute GET request.

        Args:
            url: Target URL
            headers: Optional request headers
            params: Optional query parameters
            **kwargs: Additional request options

        Returns:
            HTTP response wrapper

        Raises:
            NetworkError: When request fails

        """
        ...

    @abstractmethod
    def post(
        self,
        url: str,
        data: bytes | str | dict[str, object] | None = None,
        json: dict[str, object] | None = None,
        headers: dict[str, str] | None = None,
        **kwargs: object,
    ) -> HttpResponse:
        """
        Execute POST request.

        Args:
            url: Target URL
            data: Request body data
            json: JSON data to send
            headers: Optional request headers
            **kwargs: Additional request options

        Returns:
            HTTP response wrapper

        Raises:
            NetworkError: When request fails

        """
        ...

    @abstractmethod
    def download_file(
        self,
        url: str,
        file_path: Path,
        progress_callback: Callable[[int, int], None] | None = None,
    ) -> HttpResponse:
        """
        Download file with optional progress tracking.

        Args:
            url: File URL to download
            file_path: Destination file path
            progress_callback: Optional progress callback function

        Returns:
            HTTP response wrapper

        Raises:
            NetworkError: When download fails

        """
        ...

    @abstractmethod
    def head(
        self,
        url: str,
        headers: dict[str, str] | None = None,
        **kwargs: object,
    ) -> HttpResponse:
        """
        Execute HEAD request.

        Args:
            url: Target URL
            headers: Optional request headers
            **kwargs: Additional request options

        Returns:
            HTTP response wrapper

        Raises:
            NetworkError: When request fails

        """
        ...

    @abstractmethod
    def is_url_accessible(self, url: str) -> bool:
        """
        Check if URL is accessible without downloading content.

        Args:
            url: URL to check

        Returns:
            True if URL is accessible, False otherwise

        """
        ...

    @abstractmethod
    @contextmanager
    def session(self) -> Generator[object]:
        """
        Context manager for persistent session.

        Yields:
            HTTP client session

        """
        ...

    @abstractmethod
    def clear_cache(self) -> None:
        """Clear the response cache."""
        ...
Functions
clear_cache() abstractmethod

Clear the response cache.

Source code in src\kp_ssf_tools\core\services\http_client\interfaces.py 
144
145
146
147
@abstractmethod
def clear_cache(self) -> None:
    """Clear the response cache."""
    ...
download_file(url, file_path, progress_callback=None) abstractmethod

Download file with optional progress tracking.

Parameters:

Name Type Description Default
url str

File URL to download

 required
file_path Path

Destination file path

 required
progress_callback Callable[[int, int], None] | None

Optional progress callback function

 None

Returns:

Type Description
HttpResponse

HTTP response wrapper

Raises:

Type Description
NetworkError

When download fails
Source code in src\kp_ssf_tools\core\services\http_client\interfaces.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
@abstractmethod
def download_file(
    self,
    url: str,
    file_path: Path,
    progress_callback: Callable[[int, int], None] | None = None,
) -> HttpResponse:
    """
    Download file with optional progress tracking.

    Args:
        url: File URL to download
        file_path: Destination file path
        progress_callback: Optional progress callback function

    Returns:
        HTTP response wrapper

    Raises:
        NetworkError: When download fails

    """
    ...
get(url, headers=None, params=None, **kwargs) abstractmethod

Execute GET request.

Parameters:

Name Type Description Default
url str

Target URL

 required
headers dict[str, str] | None

Optional request headers

 None
params dict[str, str] | None

Optional query parameters

 None
**kwargs object

Additional request options

 {}

Returns:

Type Description
HttpResponse

HTTP response wrapper

Raises:

Type Description
NetworkError

When request fails
Source code in src\kp_ssf_tools\core\services\http_client\interfaces.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
@abstractmethod
def get(
    self,
    url: str,
    headers: dict[str, str] | None = None,
    params: dict[str, str] | None = None,
    **kwargs: object,
) -> HttpResponse:
    """
    Execute GET request.

    Args:
        url: Target URL
        headers: Optional request headers
        params: Optional query parameters
        **kwargs: Additional request options

    Returns:
        HTTP response wrapper

    Raises:
        NetworkError: When request fails

    """
    ...
head(url, headers=None, **kwargs) abstractmethod

Execute HEAD request.

Parameters:

Name Type Description Default
url str

Target URL

 required
headers dict[str, str] | None

Optional request headers

 None
**kwargs object

Additional request options

 {}

Returns:

Type Description
HttpResponse

HTTP response wrapper

Raises:

Type Description
NetworkError

When request fails
Source code in src\kp_ssf_tools\core\services\http_client\interfaces.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@abstractmethod
def head(
    self,
    url: str,
    headers: dict[str, str] | None = None,
    **kwargs: object,
) -> HttpResponse:
    """
    Execute HEAD request.

    Args:
        url: Target URL
        headers: Optional request headers
        **kwargs: Additional request options

    Returns:
        HTTP response wrapper

    Raises:
        NetworkError: When request fails

    """
    ...
is_url_accessible(url) abstractmethod

Check if URL is accessible without downloading content.

Parameters:

Name Type Description Default
url str

URL to check

 required

Returns:

Type Description
bool

True if URL is accessible, False otherwise
Source code in src\kp_ssf_tools\core\services\http_client\interfaces.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
@abstractmethod
def is_url_accessible(self, url: str) -> bool:
    """
    Check if URL is accessible without downloading content.

    Args:
        url: URL to check

    Returns:
        True if URL is accessible, False otherwise

    """
    ...
post(url, data=None, json=None, headers=None, **kwargs) abstractmethod

Execute POST request.

Parameters:

Name Type Description Default
url str

Target URL

 required
data bytes | str | dict[str, object] | None

Request body data

 None
json dict[str, object] | None

JSON data to send

 None
headers dict[str, str] | None

Optional request headers

 None
**kwargs object

Additional request options

 {}

Returns:

Type Description
HttpResponse

HTTP response wrapper

Raises:

Type Description
NetworkError

When request fails
Source code in src\kp_ssf_tools\core\services\http_client\interfaces.py 
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
@abstractmethod
def post(
    self,
    url: str,
    data: bytes | str | dict[str, object] | None = None,
    json: dict[str, object] | None = None,
    headers: dict[str, str] | None = None,
    **kwargs: object,
) -> HttpResponse:
    """
    Execute POST request.

    Args:
        url: Target URL
        data: Request body data
        json: JSON data to send
        headers: Optional request headers
        **kwargs: Additional request options

    Returns:
        HTTP response wrapper

    Raises:
        NetworkError: When request fails

    """
    ...
session() abstractmethod

Context manager for persistent session.

Yields:

Type Description
Generator[object]

HTTP client session
Source code in src\kp_ssf_tools\core\services\http_client\interfaces.py 
132
133
134
135
136
137
138
139
140
141
142
@abstractmethod
@contextmanager
def session(self) -> Generator[object]:
    """
    Context manager for persistent session.

    Yields:
        HTTP client session

    """
    ...

Error Handling

kp_ssf_tools.core.services.http_client.models.NetworkError

Bases: Exception

Base exception for network-related errors.

Source code in src\kp_ssf_tools\core\services\http_client\models.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
class NetworkError(Exception):
    """Base exception for network-related errors."""

    def __init__(
        self,
        message: str,
        *,
        url: str | None = None,
        status_code: int | None = None,
        retry_attempted: bool = False,
    ) -> None:
        super().__init__(message)
        self.url = url
        self.status_code = status_code
        self.retry_attempted = retry_attempted

kp_ssf_tools.core.services.http_client.models.HttpTimeoutError

Bases: NetworkError

Request timeout error.

Source code in src\kp_ssf_tools\core\services\http_client\models.py 
104
105
class HttpTimeoutError(NetworkError):
    """Request timeout error."""

kp_ssf_tools.core.services.http_client.models.HttpConnectionError

Bases: NetworkError

Connection-related error.

Source code in src\kp_ssf_tools\core\services\http_client\models.py 
108
109
class HttpConnectionError(NetworkError):
    """Connection-related error."""

kp_ssf_tools.core.services.http_client.models.HttpError

Bases: NetworkError

HTTP protocol error.

Source code in src\kp_ssf_tools\core\services\http_client\models.py 
112
113
class HttpError(NetworkError):
    """HTTP protocol error."""

Usage Patterns

Basic Dependency Injection Pattern

The HTTP Client Service follows standard dependency injection patterns:

class DataProcessor:
    def __init__(self, http_client: HttpClientProtocol):
        self._http_client = http_client

    def fetch_external_data(self, url: str) -> ProcessResult:
        response = self._http_client.get(url)
        # Process response...
        return ProcessResult(data=response.text)

Configuration-Driven Behavior

The service supports flexible configuration through the container system:

# Container configuration example
http_config = HttpConfig(
    timeout_seconds=30.0,
    max_retries=5,
    retry_strategy=RetryStrategy.EXPONENTIAL_BACKOFF,
    cache_enabled=True
)

Error Handling Strategy

All HTTP operations use consistent error handling:

try:
    response = http_client.get(url)
    return response.text
except NetworkError as e:
    # Handle network-level errors
    logger.error(f"Network operation failed: {e}")
except HttpError as e:
    # Handle HTTP protocol errors
    logger.error(f"HTTP error {e.status_code}: {e}")

Container Configuration

The HTTP Client Service integrates with the core container system for dependency injection:

Service Registration

class CoreContainer(containers.DeclarativeContainer):
    # Configuration
    http_config = providers.Configuration()

    # Services
    http_client = providers.Singleton(
        HttpClientService,
        config=http_config,
        rich_output=rich_output_service
    )

    # Dependent services
    data_processor = providers.Factory(
        DataProcessor,
        http_client=http_client
    )

Configuration Structure

# ssf-tools-config.yaml
http:
  timeout_seconds: 30.0
  max_retries: 5
  retry_strategy: "exponential_backoff"
  cache_enabled: true
  cache_ttl_seconds: 3600
  user_agent: "SSF-Tools/1.0"

Performance Considerations

Connection Management

  • Connection pooling: Reuses connections across requests
  • Resource cleanup: Automatic cleanup of network resources
  • Timeout handling: Configurable timeouts prevent hanging operations

Caching Strategy

  • Response caching: Reduces redundant network requests
  • TTL-based expiration: Ensures data freshness
  • Memory efficiency: Bounded cache size prevents memory leaks

Retry Logic

  • Exponential backoff: Prevents server overload during retries
  • Transient error detection: Distinguishes retry-able from permanent failures
  • Configurable strategies: Supports different retry patterns per use case

The HTTP Client Service provides a robust, configurable foundation for network operations across SSF Tools, emphasizing consistency, testability, and performance through proper architectural patterns.