Imbi API Client

The imbi_common.api subpackage provides an async HTTP client for talking to the Imbi API and a pydantic_settings.BaseSettings class for configuring it from environment variables.

It is used by services that integrate with Imbi (e.g. imbi-gateway) to record releases, push deployment events, patch project metadata, and look up users by external identity.

Overview

imbi_common.api.client.Imbi is a thin subclass of httpx.AsyncClient that:

• Sets the Authorization and User-Agent headers on every request.
• Adds task-oriented helpers for the bookkeeping endpoints (patch_project, find_user_by_identity, create_release, record_deployment).
• Logs non-2xx responses at warning level while still returning the raw response so callers can decide how to react.

imbi_common.api.settings.Settings loads configuration from environment variables prefixed with IMBI_CLIENT_. The expected variables are:

Field	Environment variable	Default
api_base_url	IMBI_CLIENT_API_BASE_URL	http://imbi-api:8000
api_token	IMBI_CLIENT_API_TOKEN	required
user_agent	IMBI_CLIENT_USER_AGENT	None (use the default imbi-common/{version})

Basic Usage

from imbi_common.api import Imbi, Settings

settings = Settings()

async with Imbi(
    base_url=str(settings.api_base_url),
    token=settings.api_token.get_secret_value(),
    user_agent=settings.user_agent,
) as client:
    response = await client.create_release(
        'my-org',
        'my-project',
        {'version': '1.2.3', 'title': 'v1.2.3'},
    )
    response.raise_for_status()

The constructor parameters mirror what most services want to override explicitly — base URL, bearer token, optional user-agent, optional timeout. Settings is provided for the common case of pulling those values from the environment, but callers are free to build an Imbi client from any other source.

Status-Code Conventions

Two endpoints have idempotency conventions baked into the helpers:

• create_release treats 409 Conflict as success — the release already exists — and does not log a warning. Other non-2xx responses are logged.
• record_deployment treats 404 Not Found as a non-fatal "release missing" condition and does not log a warning. Other non-2xx responses are logged.

In both cases the raw httpx.Response is returned so the caller can distinguish these states from a fully successful 2xx.

API Reference

Client

Imbi

Imbi(
    *,
    base_url: str,
    token: str,
    user_agent: str | None = None,
    timeout: float | None = None,
)

Bases: AsyncClient

Async HTTP client for the Imbi API.

Wraps httpx.AsyncClient with the Authorization and User-Agent headers wired up and adds high-level methods for the bookkeeping endpoints that integrating services call.

Parameters:

Name	Type	Description	Default
base_url	str	Root URL of the Imbi API (e.g. http://imbi-api:8000).	required
token	str	Bearer token sent on every request.	required
user_agent	str | None	Optional User-Agent header value. Defaults to imbi-common/{version} when not supplied.	None
timeout	float | None	Optional request timeout in seconds. When omitted, httpx's default applies.	None

Source code in src/imbi_common/api/client.py

def __init__(
    self,
    *,
    base_url: str,
    token: str,
    user_agent: str | None = None,
    timeout: float | None = None,
) -> None:
    headers = {
        'authorization': f'Bearer {token}',
        'user-agent': user_agent or _default_user_agent(),
    }
    kwargs: dict[str, typing.Any] = {
        'base_url': base_url,
        'headers': headers,
    }
    if timeout is not None:
        kwargs['timeout'] = timeout
    super().__init__(**kwargs)

create_release async

create_release(
    org_slug: str,
    project_id: str,
    body: Mapping[str, object],
) -> httpx.Response

POST a new release to the project's releases collection.

Parameters:

Name	Type	Description	Default
org_slug	str	Organization slug.	required
project_id	str	Project identifier.	required
body	Mapping[str, object]	Release payload (version, title, optional created_by, etc.).	required

Returns:

Type	Description
Response	The raw httpx.Response. 409 Conflict is treated as
Response	an idempotent success and is not logged; other non-2xx
Response	responses are logged at warning level.

Source code in src/imbi_common/api/client.py

async def create_release(
    self,
    org_slug: str,
    project_id: str,
    body: abc.Mapping[str, object],
) -> httpx.Response:
    """POST a new release to the project's releases collection.

    Args:
        org_slug: Organization slug.
        project_id: Project identifier.
        body: Release payload (`version`, `title`, optional
            `created_by`, etc.).

    Returns:
        The raw `httpx.Response`. `409 Conflict` is treated as
        an idempotent success and is not logged; other non-2xx
        responses are logged at warning level.
    """
    url = f'/organizations/{org_slug}/projects/{project_id}/releases/'
    LOGGER.debug('Creating release %s', url)
    response = await self.post(url, json=dict(body))
    if response.is_error and (
        response.status_code != http.HTTPStatus.CONFLICT
    ):
        LOGGER.warning(
            'Failed to create release %r: %s', url, response.text
        )
    return response

find_user_by_identity async

find_user_by_identity(
    plugin_slug: str, subject: str
) -> str | None

Look up an Imbi user by external identity subject.

Parameters:

Name	Type	Description	Default
plugin_slug	str	Slug of the identity plugin (e.g. github).	required
subject	str	The plugin-specific subject identifier.	required

Returns:

Type	Description
str | None	The user's email — the principal identity used by the
str | None	Release created_by field — or None when no active
str | None	identity connection matches or the lookup fails. Transport
str | None	errors and malformed JSON responses are also coerced to
str | None	None with a warning log.

Source code in src/imbi_common/api/client.py

async def find_user_by_identity(
    self, plugin_slug: str, subject: str
) -> str | None:
    """Look up an Imbi user by external identity subject.

    Args:
        plugin_slug: Slug of the identity plugin (e.g. `github`).
        subject: The plugin-specific subject identifier.

    Returns:
        The user's email — the principal identity used by the
        Release `created_by` field — or `None` when no active
        identity connection matches or the lookup fails. Transport
        errors and malformed JSON responses are also coerced to
        `None` with a warning log.
    """
    try:
        response = await self.get(
            '/users/by-identity',
            params={'plugin_slug': plugin_slug, 'subject': subject},
        )
    except httpx.RequestError as exc:
        LOGGER.warning(
            'Failed to look up user for plugin=%r subject=%r: %s',
            plugin_slug,
            subject,
            exc,
        )
        return None
    if response.status_code == http.HTTPStatus.NOT_FOUND:
        return None
    if response.is_error:
        LOGGER.warning(
            'Failed to look up user for plugin=%r subject=%r: %s',
            plugin_slug,
            subject,
            response.text,
        )
        return None
    try:
        data = response.json()
    except ValueError:
        LOGGER.warning(
            'Failed to decode user lookup response for '
            'plugin=%r subject=%r',
            plugin_slug,
            subject,
        )
        return None
    email = data.get('email') if isinstance(data, dict) else None
    return str(email) if email else None

patch_project async

patch_project(
    org_slug: str,
    project_id: str,
    patch: Iterable[Mapping[str, object]],
) -> httpx.Response

Send a JSON Patch to a project.

Parameters:

Name	Type	Description	Default
org_slug	str	Organization slug.	required
project_id	str	Project identifier.	required
patch	Iterable[Mapping[str, object]]	JSON Patch operations.	required

Returns:

Type	Description
Response	The raw httpx.Response. Non-2xx responses are logged at
Response	warning level and returned to the caller.

Source code in src/imbi_common/api/client.py

async def patch_project(
    self,
    org_slug: str,
    project_id: str,
    patch: abc.Iterable[abc.Mapping[str, object]],
) -> httpx.Response:
    """Send a JSON Patch to a project.

    Args:
        org_slug: Organization slug.
        project_id: Project identifier.
        patch: JSON Patch operations.

    Returns:
        The raw `httpx.Response`. Non-2xx responses are logged at
        warning level and returned to the caller.
    """
    url = f'/organizations/{org_slug}/projects/{project_id}'
    LOGGER.debug('Patching project %s', url)
    response = await self.patch(url, json=list(patch))
    if response.is_error:
        try:
            detail: object = response.json()
        except ValueError:
            detail = response.content
        LOGGER.warning('Failed to patch project %r: %r', url, detail)
    return response

record_deployment async

record_deployment(
    org_slug: str,
    project_id: str,
    version: str,
    env_slug: str,
    body: Mapping[str, object],
) -> httpx.Response

POST a deployment event onto a release's environment edge.

Parameters:

Name	Type	Description	Default
org_slug	str	Organization slug.	required
project_id	str	Project identifier.	required
version	str	Release version string.	required
env_slug	str	Target environment slug.	required
body	Mapping[str, object]	Deployment event payload (status, optional note, etc.).	required

Returns:

Type	Description
Response	The raw httpx.Response. 404 Not Found is treated
Response	as a non-fatal "release missing" condition and is not
Response	logged; other non-2xx responses are logged at warning
Response	level.

Source code in src/imbi_common/api/client.py

async def record_deployment(
    self,
    org_slug: str,
    project_id: str,
    version: str,
    env_slug: str,
    body: abc.Mapping[str, object],
) -> httpx.Response:
    """POST a deployment event onto a release's environment edge.

    Args:
        org_slug: Organization slug.
        project_id: Project identifier.
        version: Release version string.
        env_slug: Target environment slug.
        body: Deployment event payload (`status`, optional
            `note`, etc.).

    Returns:
        The raw `httpx.Response`. `404 Not Found` is treated
        as a non-fatal "release missing" condition and is not
        logged; other non-2xx responses are logged at warning
        level.
    """
    url = (
        f'/organizations/{org_slug}/projects/{project_id}'
        f'/releases/{version}/environments/{env_slug}'
    )
    LOGGER.debug('Recording deployment %s', url)
    response = await self.post(url, json=dict(body))
    if response.is_error and (
        response.status_code != http.HTTPStatus.NOT_FOUND
    ):
        LOGGER.warning(
            'Failed to record deployment %r: %s', url, response.text
        )
    return response

Settings

Settings

Bases: BaseSettings

Configuration for imbi_common.api.client.Imbi.

Values are loaded from environment variables prefixed with IMBI_CLIENT_:

• IMBI_CLIENT_API_BASE_URL — root URL of the Imbi API. Defaults to http://imbi-api:8000.
• IMBI_CLIENT_API_TOKEN — bearer token sent on every request. Required; instantiation raises ValidationError when unset.
• IMBI_CLIENT_USER_AGENT — optional User-Agent override. When unset, the client falls back to imbi-common/{version}.