Skip to content

http_service

appimage_updater.core.http_service

HTTP service with optional tracing support and connection pooling.

AsyncClient(**kwargs)

HTTP client context manager that uses the global client.

Source code in src/appimage_updater/core/http_service.py
def __init__(self, **kwargs: Any):
    self.kwargs = kwargs
    self._client: TracingAsyncClient | None = None

kwargs = kwargs instance-attribute

GlobalHTTPClientImpl()

Global HTTP client manager with connection pooling.

Source code in src/appimage_updater/core/http_service.py
def __init__(self) -> None:
    self._client: httpx.AsyncClient | None = None
    self._tracer: Any | None = None
    self._initialized = False

close() async

Close the global HTTP client.

Source code in src/appimage_updater/core/http_service.py
async def close(self) -> None:
    """Close the global HTTP client."""
    if self._client:
        await self._client.aclose()
        self._client = None

get_client(**kwargs) async

Get the global HTTP client with tracing.

Source code in src/appimage_updater/core/http_service.py
async def get_client(self, **kwargs: Any) -> TracingAsyncClient:
    """Get the global HTTP client with tracing."""
    client = await self._ensure_client(**kwargs)
    return TracingAsyncClient(client, self._tracer)

set_tracer(tracer)

Set the global tracer.

Source code in src/appimage_updater/core/http_service.py
def set_tracer(self, tracer: Any | None) -> None:
    """Set the global tracer."""
    self._tracer = tracer

TracingAsyncClient(client, tracer=None)

HTTP client wrapper with optional tracing.

Source code in src/appimage_updater/core/http_service.py
def __init__(self, client: httpx.AsyncClient, tracer: Any | None = None):
    self._client = client
    self._tracer = tracer

delete(url, **kwargs) async

DELETE request with optional tracing.

Source code in src/appimage_updater/core/http_service.py
async def delete(self, url: str, **kwargs: Any) -> Any:
    """DELETE request with optional tracing."""
    return await self._traced_request("DELETE", url, self._client.delete, url, **kwargs)

get(url, **kwargs) async

GET request with optional tracing.

Source code in src/appimage_updater/core/http_service.py
async def get(self, url: str, **kwargs: Any) -> Any:
    """GET request with optional tracing."""
    return await self._traced_request("GET", url, self._client.get, url, **kwargs)

post(url, **kwargs) async

POST request with optional tracing.

Source code in src/appimage_updater/core/http_service.py
async def post(self, url: str, **kwargs: Any) -> Any:
    """POST request with optional tracing."""
    return await self._traced_request("POST", url, self._client.post, url, **kwargs)

put(url, **kwargs) async

PUT request with optional tracing.

Source code in src/appimage_updater/core/http_service.py
async def put(self, url: str, **kwargs: Any) -> Any:
    """PUT request with optional tracing."""
    return await self._traced_request("PUT", url, self._client.put, url, **kwargs)

GlobalHTTPClient() cached

Singleton HTTP client manager factory.

Source code in src/appimage_updater/core/http_service.py
@lru_cache(maxsize=1)
def GlobalHTTPClient() -> GlobalHTTPClientImpl:  # noqa: N802
    """Singleton HTTP client manager factory."""
    return GlobalHTTPClientImpl()

disable_global_trace()

Disable global HTTP tracing.

Source code in src/appimage_updater/core/http_service.py
def disable_global_trace() -> None:
    """Disable global HTTP tracing."""

    tracer = getHTTPTrace()
    if tracer.enabled and tracer.output_formatter:
        tracer.output_formatter.print_message("HTTP TRACE: Stopping request tracking")
    tracer.enabled = False

    global_client = GlobalHTTPClient()
    global_client.set_tracer(None)

enable_global_trace(output_formatter=None)

Enable global HTTP tracing.

Parameters:

Name Type Description Default
output_formatter Any

Output formatter to use for trace messages

None
Source code in src/appimage_updater/core/http_service.py
def enable_global_trace(output_formatter: Any = None) -> None:
    """Enable global HTTP tracing.

    Args:
        output_formatter: Output formatter to use for trace messages
    """
    tracer = getHTTPTrace(output_formatter)
    tracer.enabled = True
    if output_formatter:
        output_formatter.print_message("HTTP TRACE: Starting request tracking")

    global_client = GlobalHTTPClient()
    global_client.set_tracer(tracer)

get_http_client(**kwargs)

Get HTTP client with global tracing configuration.

Parameters:

Name Type Description Default
**kwargs Any

HTTP client parameters (timeout, follow_redirects, etc.)

{}

Returns:

Type Description
Any

AsyncClient instance (or mock client for testing) that uses the global client with connection pooling

Source code in src/appimage_updater/core/http_service.py
def get_http_client(**kwargs: Any) -> Any:
    """Get HTTP client with global tracing configuration.

    Args:
        **kwargs: HTTP client parameters (timeout, follow_redirects, etc.)

    Returns:
        AsyncClient instance (or mock client for testing) that uses the global client with connection pooling
    """
    # Use injected factory if available (for testing)
    if _http_client_factory is not None:
        return _http_client_factory(**kwargs)

    # Otherwise use the real AsyncClient
    return AsyncClient(**kwargs)

reset_http_client_factory()

Reset HTTP client factory to default (production) behavior.

Source code in src/appimage_updater/core/http_service.py
def reset_http_client_factory() -> None:
    """Reset HTTP client factory to default (production) behavior."""
    global _http_client_factory
    _http_client_factory = None

set_http_client_factory(factory)

Set custom HTTP client factory (mainly for testing).

Parameters:

Name Type Description Default
factory Callable[..., Any] | None

A callable that returns an HTTP client, or None to reset to default

required
Source code in src/appimage_updater/core/http_service.py
def set_http_client_factory(factory: Callable[..., Any] | None) -> None:
    """Set custom HTTP client factory (mainly for testing).

    Args:
        factory: A callable that returns an HTTP client, or None to reset to default
    """
    global _http_client_factory
    _http_client_factory = factory