Base

provide.foundation.transport.base

TODO: Add module docstring.

Classes

Request

Protocol-agnostic request.

Attributes

base_url property

base_url: str

Extract base URL from URI.

transport_type property

transport_type: TransportType

Infer transport type from URI scheme.

Response

Protocol-agnostic response.

Attributes

text property

text: str

Get response body as text.

Functions

is_success

is_success() -> bool

Check if response indicates success.

Source code in provide/foundation/transport/base.py

def is_success(self) -> bool:
    """Check if response indicates success."""
    return 200 <= self.status < 300

json

json() -> Any

Parse response body as JSON.

Source code in provide/foundation/transport/base.py

def json(self) -> Any:
    """Parse response body as JSON."""
    if isinstance(self.body, bytes):
        return json_loads(self.body.decode("utf-8"))
    if isinstance(self.body, str):
        return json_loads(self.body)
    raise ValueError("Response body is not JSON-parseable")

raise_for_status

raise_for_status() -> None

Raise error if response status indicates failure.

Source code in provide/foundation/transport/base.py

def raise_for_status(self) -> None:
    """Raise error if response status indicates failure."""
    if not self.is_success():
        from provide.foundation.transport.errors import HTTPResponseError

        raise HTTPResponseError(
            f"Request failed with status {self.status}",
            status_code=self.status,
            response=self,
        )

Transport

Bases: Protocol

Abstract transport protocol.

Functions

__aenter__ async

__aenter__() -> Transport

Context manager entry.

Source code in provide/foundation/transport/base.py

async def __aenter__(self) -> Transport:
    """Context manager entry."""
    await self.connect()
    return self

__aexit__ async

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: Any,
) -> None

Context manager exit.

Source code in provide/foundation/transport/base.py

async def __aexit__(
    self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any
) -> None:
    """Context manager exit."""
    await self.disconnect()

connect async

connect() -> None

Establish connection if needed.

Source code in provide/foundation/transport/base.py

async def connect(self) -> None:
    """Establish connection if needed."""
    ...

disconnect async

disconnect() -> None

Close connection if needed.

Source code in provide/foundation/transport/base.py

async def disconnect(self) -> None:
    """Close connection if needed."""
    ...

execute async

execute(request: Request) -> Response

Execute a request and return response.

Source code in provide/foundation/transport/base.py

async def execute(self, request: Request) -> Response:
    """Execute a request and return response."""
    ...

stream async

stream(request: Request) -> AsyncIterator[bytes]

Stream response data.

Source code in provide/foundation/transport/base.py

async def stream(self, request: Request) -> AsyncIterator[bytes]:
    """Stream response data."""
    ...

supports

supports(transport_type: TransportType) -> bool

Check if this transport handles the given type.

Source code in provide/foundation/transport/base.py

def supports(self, transport_type: TransportType) -> bool:
    """Check if this transport handles the given type."""
    ...

TransportBase

TransportBase()

Bases: ABC

Base class for transport implementations.

Source code in provide/foundation/transport/base.py

def __init__(self) -> None:
    self._logger = get_logger(self.__class__.__name__)

Functions

connect async

connect() -> None

Default connect implementation.

Source code in provide/foundation/transport/base.py

async def connect(self) -> None:
    """Default connect implementation."""
    self._logger.trace("Transport connecting")

disconnect async

disconnect() -> None

Default disconnect implementation.

Source code in provide/foundation/transport/base.py

async def disconnect(self) -> None:
    """Default disconnect implementation."""
    self._logger.trace("Transport disconnecting")

execute abstractmethod async

execute(request: Request) -> Response

Execute a request and return response.

Source code in provide/foundation/transport/base.py

@abstractmethod
async def execute(self, request: Request) -> Response:
    """Execute a request and return response."""

stream async

stream(request: Request) -> AsyncIterator[bytes]

Stream response data incrementally.

Note: This is an intentional design limitation. The base Transport class does not implement streaming. Subclasses may override this method to provide streaming support if needed for specific use cases.

Raises:

Type	Description
NotImplementedError	Streaming is not supported by this transport implementation

Source code in provide/foundation/transport/base.py

async def stream(self, request: Request) -> AsyncIterator[bytes]:
    """Stream response data incrementally.

    Note: This is an intentional design limitation. The base Transport class
    does not implement streaming. Subclasses may override this method to provide
    streaming support if needed for specific use cases.

    Raises:
        NotImplementedError: Streaming is not supported by this transport implementation
    """
    raise NotImplementedError(
        f"{self.__class__.__name__} does not support streaming. "
        f"Override this method in a subclass to implement streaming support."
    )

supports abstractmethod

supports(transport_type: TransportType) -> bool

Check if this transport handles the given type.

Source code in provide/foundation/transport/base.py

@abstractmethod
def supports(self, transport_type: TransportType) -> bool:
    """Check if this transport handles the given type."""

Functions