API Client

client

T3 API client built on httpx.

Provides :class:T3APIClient, an async httpx-based client that handles authentication, retries, and response parsing for the T3 API.

T3APIClient

T3APIClient(*, config: Optional[HTTPConfig] = None, retry_policy: Optional[RetryPolicy] = None, logging_hooks: Optional[LoggingHooks] = None, headers: Optional[Dict[str, str]] = None)

Async T3 API client using httpx.AsyncClient.

This client provides a high-level interface to the T3 API, handling authentication, retries, and response parsing with async/await support for better performance when making many concurrent requests.

Initialize the async T3 API client.

Parameters:

Name	Type	Description	Default
config	Optional[HTTPConfig]	HTTP configuration (host, timeout, etc.)	None
retry_policy	Optional[RetryPolicy]	Retry policy for failed requests	None
logging_hooks	Optional[LoggingHooks]	Optional request/response logging	None
headers	Optional[Dict[str, str]]	Additional headers to include with requests	None

Source code in t3api_utils/api/client.py

def __init__(
    self,
    *,
    config: Optional[HTTPConfig] = None,
    retry_policy: Optional[RetryPolicy] = None,
    logging_hooks: Optional[LoggingHooks] = None,
    headers: Optional[Dict[str, str]] = None,
) -> None:
    """Initialize the async T3 API client.

    Args:
        config: HTTP configuration (host, timeout, etc.)
        retry_policy: Retry policy for failed requests
        logging_hooks: Optional request/response logging
        headers: Additional headers to include with requests
    """
    self._config = config or HTTPConfig()
    self._retry_policy = retry_policy or RetryPolicy()
    self._logging_hooks = logging_hooks
    self._extra_headers = headers or {}

    # Build the async httpx client
    self._client = build_async_client(
        config=self._config,
        headers=self._extra_headers,
        hooks=self._logging_hooks,
    )

    # Track authentication state
    self._authenticated = False
    self._access_token: Optional[str] = None

is_authenticated property

is_authenticated: bool

Check if client is authenticated.

Returns:

Type	Description
bool	True if an access token has been set and authentication succeeded.

access_token property

access_token: Optional[str]

Get the current access token.

Returns:

Type	Description
Optional[str]	The JWT bearer token string, or None if unauthenticated.

__aenter__ async

__aenter__() -> T3APIClient

Async context manager entry.

Source code in t3api_utils/api/client.py

async def __aenter__(self) -> T3APIClient:
    """Async context manager entry."""
    return self

__aexit__ async

__aexit__(exc_type: object, exc_val: object, exc_tb: object) -> None

Async context manager exit.

Source code in t3api_utils/api/client.py

async def __aexit__(self, exc_type: object, exc_val: object, exc_tb: object) -> None:
    """Async context manager exit."""
    await self.close()

close async

close() -> None

Close the underlying HTTP client.

Source code in t3api_utils/api/client.py

async def close(self) -> None:
    """Close the underlying HTTP client."""
    await self._client.aclose()

set_access_token

set_access_token(token: str) -> None

Set the access token for authentication.

Parameters:

Name	Type	Description	Default
token	str	Bearer token for API authentication	required

Source code in t3api_utils/api/client.py

def set_access_token(self, token: str) -> None:
    """Set the access token for authentication.

    Args:
        token: Bearer token for API authentication
    """
    self._access_token = token
    set_bearer_token(client=self._client, token=token)
    self._authenticated = True

clear_access_token

clear_access_token() -> None

Clear the access token and mark the client as unauthenticated.

Removes the Authorization header from the underlying httpx client.

Source code in t3api_utils/api/client.py

def clear_access_token(self) -> None:
    """Clear the access token and mark the client as unauthenticated.

    Removes the ``Authorization`` header from the underlying httpx client.
    """
    self._access_token = None
    clear_bearer_token(client=self._client)
    self._authenticated = False

authenticate_with_credentials async

authenticate_with_credentials(*, hostname: str, username: str, password: str, otp: Optional[str] = None, email: Optional[str] = None) -> AuthResponseData

Authenticate with T3 API using credentials.

Parameters:

Name	Type	Description	Default
hostname	str	The hostname for authentication	required
username	str	Username	required
password	str	Password	required
otp	Optional[str]	Optional one-time password	None
email	Optional[str]	Optional email address	None

Returns:

Type	Description
AuthResponseData	AuthResponseData containing access token and metadata

Raises:

Type	Description
T3HTTPError	If authentication fails

Source code in t3api_utils/api/client.py

async def authenticate_with_credentials(
    self,
    *,
    hostname: str,
    username: str,
    password: str,
    otp: Optional[str] = None,
    email: Optional[str] = None,
) -> AuthResponseData:
    """Authenticate with T3 API using credentials.

    Args:
        hostname: The hostname for authentication
        username: Username
        password: Password
        otp: Optional one-time password
        email: Optional email address

    Returns:
        AuthResponseData containing access token and metadata

    Raises:
        T3HTTPError: If authentication fails
    """
    # Prepare request payload
    payload = {
        "hostname": hostname,
        "username": username,
        "password": password,
    }

    if otp is not None:
        payload["otp"] = otp

    if email is not None:
        payload["email"] = email

    # Make the request
    try:
        response_data = await arequest_json(
            aclient=self._client,
            method="POST",
            url="/v2/auth/credentials",
            json_body=payload,
            policy=self._retry_policy,
            expected_status=200,
        )

        # Set the token for future requests
        self.set_access_token(response_data["accessToken"])

        return cast(AuthResponseData, response_data)

    except T3HTTPError as e:
        # Re-raise with more context
        raise T3HTTPError(f"Authentication failed: {e}", response=e.response) from e

authenticate_with_api_key async

authenticate_with_api_key(*, api_key: str, state_code: str) -> AuthResponseData

Authenticate with T3 API using API key.

Parameters:

Name	Type	Description	Default
api_key	str	API key for the T3 API	required
state_code	str	State code (e.g., "CA", "MO", "CO", "MI")	required

Returns:

Type	Description
AuthResponseData	AuthResponseData containing access token and metadata

Raises:

Type	Description
T3HTTPError	If authentication fails

Source code in t3api_utils/api/client.py

async def authenticate_with_api_key(
    self,
    *,
    api_key: str,
    state_code: str,
) -> AuthResponseData:
    """Authenticate with T3 API using API key.

    Args:
        api_key: API key for the T3 API
        state_code: State code (e.g., "CA", "MO", "CO", "MI")

    Returns:
        AuthResponseData containing access token and metadata

    Raises:
        T3HTTPError: If authentication fails
    """
    # Prepare request payload
    payload = {
        "apiKey": api_key,
        "stateCode": state_code,
    }

    # Make the request
    try:
        response_data = await arequest_json(
            aclient=self._client,
            method="POST",
            url="/v2/auth/apikey",
            json_body=payload,
            policy=self._retry_policy,
            expected_status=200,
        )

        # Set the token for future requests
        self.set_access_token(response_data["accessToken"])

        return cast(AuthResponseData, response_data)

    except T3HTTPError as e:
        # Re-raise with more context
        raise T3HTTPError(f"API key authentication failed: {e}", response=e.response) from e