honeyhive.api.client

HoneyHive API Client.

This module provides the main HoneyHive client with an ergonomic interface wrapping the auto-generated API code.

Usage::

from honeyhive.api import HoneyHive

client = HoneyHive(api_key="your-api-key")

# Sync usage
configs = client.configurations.list()

# Async usage
configs = await client.configurations.list_async()

logger module-attribute

logger = logging.getLogger(__name__)

QUERY_BATCH_SIZE module-attribute

QUERY_BATCH_SIZE = 100

EXPORT_TIMEOUT module-attribute

EXPORT_TIMEOUT = _build_export_timeout()

T module-attribute

T = TypeVar('T')

ChartsAPI

Bases: BaseAPI

Charts API.

Every method takes an optional project_id. With it, the request names the project in its path and is sent with the data plane API key. Without it, the request acts on the project API key's project, is sent with that key, and issues a DeprecationWarning.

Each public method warns and then calls its private core, which sends the request and never warns. SDK code that reaches a chart operation calls the core, so the warning reaches only a caller who can pass project_id.

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

class ChartsAPI(BaseAPI):
    """Charts API.

    Every method takes an optional ``project_id``. With it, the request names
    the project in its path and is sent with the data plane API key. Without
    it, the request acts on the project API key's project, is sent with that
    key, and issues a ``DeprecationWarning``.

    Each public method warns and then calls its private core, which sends the
    request and never warns. SDK code that reaches a chart operation calls the
    core, so the warning reaches only a caller who can pass ``project_id``.
    """

    def list(self, *, project_id: Optional[str] = None) -> GetChartsResponse:
        """List the charts in a project.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            project_id: The project whose charts are listed. Omitting it lists
                the project API key's charts, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("charts", "list")
        return self._list(project_id=project_id)

    def get(
        self, chart_id: str, *, project_id: Optional[str] = None
    ) -> GetChartResponse:
        """Get a chart by ID.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            chart_id: The chart to get.
            project_id: The project that owns the chart. Omitting it reads from
                the project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("charts", "get")
        return self._get(chart_id, project_id=project_id)

    def create(
        self, request: CreateChartRequest, *, project_id: Optional[str] = None
    ) -> CreateChartResponse:
        """Create a chart.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            request: The chart to create.
            project_id: The project the chart is created in. Omitting it creates
                it in the project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("charts", "create")
        return self._create(request, project_id=project_id)

    def update(
        self,
        chart_id: str,
        request: UpdateChartRequest,
        *,
        project_id: Optional[str] = None,
    ) -> UpdateChartResponse:
        """Update a chart. Only the fields set in ``request`` are changed.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            chart_id: The chart to update.
            request: The fields to change.
            project_id: The project that owns the chart. Omitting it updates it
                in the project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("charts", "update")
        return self._update(chart_id, request, project_id=project_id)

    def delete(
        self, chart_id: str, *, project_id: Optional[str] = None
    ) -> DeleteChartResponse:
        """Delete a chart and remove it from the project's dashboards.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            chart_id: The chart to delete.
            project_id: The project that owns the chart. Omitting it deletes it
                from the project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("charts", "delete")
        return self._delete(chart_id, project_id=project_id)

    async def list_async(
        self, *, project_id: Optional[str] = None
    ) -> GetChartsResponse:
        """List the charts in a project asynchronously. See ``list()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("charts", "list_async")
        return await self._list_async(project_id=project_id)

    async def get_async(
        self, chart_id: str, *, project_id: Optional[str] = None
    ) -> GetChartResponse:
        """Get a chart by ID asynchronously. See ``get()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("charts", "get_async")
        return await self._get_async(chart_id, project_id=project_id)

    async def create_async(
        self, request: CreateChartRequest, *, project_id: Optional[str] = None
    ) -> CreateChartResponse:
        """Create a chart asynchronously. See ``create()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("charts", "create_async")
        return await self._create_async(request, project_id=project_id)

    async def update_async(
        self,
        chart_id: str,
        request: UpdateChartRequest,
        *,
        project_id: Optional[str] = None,
    ) -> UpdateChartResponse:
        """Update a chart asynchronously. See ``update()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("charts", "update_async")
        return await self._update_async(chart_id, request, project_id=project_id)

    async def delete_async(
        self, chart_id: str, *, project_id: Optional[str] = None
    ) -> DeleteChartResponse:
        """Delete a chart asynchronously. See ``delete()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("charts", "delete_async")
        return await self._delete_async(chart_id, project_id=project_id)

    def query(
        self, request: ChartQueryRequest, *, project_id: str
    ) -> ChartQueryResponse:
        """Aggregate a project's events into one chart result.

        Accepted keys:
            Data plane fine-grained API key with ``project.event.get``

        Args:
            request: The date range, filters, and query to run. The filters
                must include an ``event_type`` filter.
            project_id: The project whose events are queried.
        """
        return charts_svc.queryChart(
            self._data_plane_api_config, project_id=project_id, data=request
        )

    async def query_async(
        self, request: ChartQueryRequest, *, project_id: str
    ) -> ChartQueryResponse:
        """Aggregate a project's events asynchronously. See ``query()``.

        Accepted keys:
            Data plane fine-grained API key with ``project.event.get``
        """
        return await charts_svc_async.queryChart(
            self._data_plane_api_config, project_id=project_id, data=request
        )

    # ---- Cores: send the request, never warn ----

    def _list(self, *, project_id: Optional[str] = None) -> GetChartsResponse:
        """The request ``list()`` sends."""
        if project_id is None:
            return charts_svc.getChartsLegacy(self._api_config)
        return charts_svc.getCharts(self._data_plane_api_config, project_id=project_id)

    def _get(
        self, chart_id: str, *, project_id: Optional[str] = None
    ) -> GetChartResponse:
        """The request ``get()`` sends."""
        if project_id is None:
            return charts_svc.getChartLegacy(self._api_config, chart_id=chart_id)
        return charts_svc.getChart(
            self._data_plane_api_config, project_id=project_id, chart_id=chart_id
        )

    def _create(
        self, request: CreateChartRequest, *, project_id: Optional[str] = None
    ) -> CreateChartResponse:
        """The request ``create()`` sends."""
        if project_id is None:
            return charts_svc.createChartLegacy(self._api_config, data=request)
        return charts_svc.createChart(
            self._data_plane_api_config, project_id=project_id, data=request
        )

    def _update(
        self,
        chart_id: str,
        request: UpdateChartRequest,
        *,
        project_id: Optional[str] = None,
    ) -> UpdateChartResponse:
        """The request ``update()`` sends."""
        if project_id is None:
            return charts_svc.updateChartLegacy(
                self._api_config, chart_id=chart_id, data=request
            )
        return charts_svc.updateChart(
            self._data_plane_api_config,
            project_id=project_id,
            chart_id=chart_id,
            data=request,
        )

    def _delete(
        self, chart_id: str, *, project_id: Optional[str] = None
    ) -> DeleteChartResponse:
        """The request ``delete()`` sends."""
        if project_id is None:
            return charts_svc.deleteChartLegacy(self._api_config, chart_id=chart_id)
        return charts_svc.deleteChart(
            self._data_plane_api_config, project_id=project_id, chart_id=chart_id
        )

    async def _list_async(
        self, *, project_id: Optional[str] = None
    ) -> GetChartsResponse:
        """The request ``list_async()`` sends."""
        if project_id is None:
            return await charts_svc_async.getChartsLegacy(self._api_config)
        return await charts_svc_async.getCharts(
            self._data_plane_api_config, project_id=project_id
        )

    async def _get_async(
        self, chart_id: str, *, project_id: Optional[str] = None
    ) -> GetChartResponse:
        """The request ``get_async()`` sends."""
        if project_id is None:
            return await charts_svc_async.getChartLegacy(
                self._api_config, chart_id=chart_id
            )
        return await charts_svc_async.getChart(
            self._data_plane_api_config, project_id=project_id, chart_id=chart_id
        )

    async def _create_async(
        self, request: CreateChartRequest, *, project_id: Optional[str] = None
    ) -> CreateChartResponse:
        """The request ``create_async()`` sends."""
        if project_id is None:
            return await charts_svc_async.createChartLegacy(
                self._api_config, data=request
            )
        return await charts_svc_async.createChart(
            self._data_plane_api_config, project_id=project_id, data=request
        )

    async def _update_async(
        self,
        chart_id: str,
        request: UpdateChartRequest,
        *,
        project_id: Optional[str] = None,
    ) -> UpdateChartResponse:
        """The request ``update_async()`` sends."""
        if project_id is None:
            return await charts_svc_async.updateChartLegacy(
                self._api_config, chart_id=chart_id, data=request
            )
        return await charts_svc_async.updateChart(
            self._data_plane_api_config,
            project_id=project_id,
            chart_id=chart_id,
            data=request,
        )

    async def _delete_async(
        self, chart_id: str, *, project_id: Optional[str] = None
    ) -> DeleteChartResponse:
        """The request ``delete_async()`` sends."""
        if project_id is None:
            return await charts_svc_async.deleteChartLegacy(
                self._api_config, chart_id=chart_id
            )
        return await charts_svc_async.deleteChart(
            self._data_plane_api_config, project_id=project_id, chart_id=chart_id
        )

list

list(
    *, project_id: Optional[str] = None
) -> GetChartsResponse

List the charts in a project.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
project_id	Optional[str]	The project whose charts are listed. Omitting it lists the project API key's charts, which is deprecated.	None

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

def list(self, *, project_id: Optional[str] = None) -> GetChartsResponse:
    """List the charts in a project.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        project_id: The project whose charts are listed. Omitting it lists
            the project API key's charts, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("charts", "list")
    return self._list(project_id=project_id)

get

get(
    chart_id: str, *, project_id: Optional[str] = None
) -> GetChartResponse

Get a chart by ID.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
chart_id	str	The chart to get.	required
project_id	Optional[str]	The project that owns the chart. Omitting it reads from the project API key's project, which is deprecated.	None

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

def get(
    self, chart_id: str, *, project_id: Optional[str] = None
) -> GetChartResponse:
    """Get a chart by ID.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        chart_id: The chart to get.
        project_id: The project that owns the chart. Omitting it reads from
            the project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("charts", "get")
    return self._get(chart_id, project_id=project_id)

create

create(
    request: CreateChartRequest,
    *,
    project_id: Optional[str] = None
) -> CreateChartResponse

Create a chart.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
request	CreateChartRequest	The chart to create.	required
project_id	Optional[str]	The project the chart is created in. Omitting it creates it in the project API key's project, which is deprecated.	None

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

def create(
    self, request: CreateChartRequest, *, project_id: Optional[str] = None
) -> CreateChartResponse:
    """Create a chart.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        request: The chart to create.
        project_id: The project the chart is created in. Omitting it creates
            it in the project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("charts", "create")
    return self._create(request, project_id=project_id)

update

update(
    chart_id: str,
    request: UpdateChartRequest,
    *,
    project_id: Optional[str] = None
) -> UpdateChartResponse

Update a chart. Only the fields set in request are changed.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
chart_id	str	The chart to update.	required
request	UpdateChartRequest	The fields to change.	required
project_id	Optional[str]	The project that owns the chart. Omitting it updates it in the project API key's project, which is deprecated.	None

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

def update(
    self,
    chart_id: str,
    request: UpdateChartRequest,
    *,
    project_id: Optional[str] = None,
) -> UpdateChartResponse:
    """Update a chart. Only the fields set in ``request`` are changed.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        chart_id: The chart to update.
        request: The fields to change.
        project_id: The project that owns the chart. Omitting it updates it
            in the project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("charts", "update")
    return self._update(chart_id, request, project_id=project_id)

delete

delete(
    chart_id: str, *, project_id: Optional[str] = None
) -> DeleteChartResponse

Delete a chart and remove it from the project's dashboards.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
chart_id	str	The chart to delete.	required
project_id	Optional[str]	The project that owns the chart. Omitting it deletes it from the project API key's project, which is deprecated.	None

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

def delete(
    self, chart_id: str, *, project_id: Optional[str] = None
) -> DeleteChartResponse:
    """Delete a chart and remove it from the project's dashboards.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        chart_id: The chart to delete.
        project_id: The project that owns the chart. Omitting it deletes it
            from the project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("charts", "delete")
    return self._delete(chart_id, project_id=project_id)

list_async async

list_async(
    *, project_id: Optional[str] = None
) -> GetChartsResponse

List the charts in a project asynchronously. See list().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def list_async(
    self, *, project_id: Optional[str] = None
) -> GetChartsResponse:
    """List the charts in a project asynchronously. See ``list()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("charts", "list_async")
    return await self._list_async(project_id=project_id)

get_async async

get_async(
    chart_id: str, *, project_id: Optional[str] = None
) -> GetChartResponse

Get a chart by ID asynchronously. See get().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def get_async(
    self, chart_id: str, *, project_id: Optional[str] = None
) -> GetChartResponse:
    """Get a chart by ID asynchronously. See ``get()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("charts", "get_async")
    return await self._get_async(chart_id, project_id=project_id)

create_async async

create_async(
    request: CreateChartRequest,
    *,
    project_id: Optional[str] = None
) -> CreateChartResponse

Create a chart asynchronously. See create().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def create_async(
    self, request: CreateChartRequest, *, project_id: Optional[str] = None
) -> CreateChartResponse:
    """Create a chart asynchronously. See ``create()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("charts", "create_async")
    return await self._create_async(request, project_id=project_id)

update_async async

update_async(
    chart_id: str,
    request: UpdateChartRequest,
    *,
    project_id: Optional[str] = None
) -> UpdateChartResponse

Update a chart asynchronously. See update().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def update_async(
    self,
    chart_id: str,
    request: UpdateChartRequest,
    *,
    project_id: Optional[str] = None,
) -> UpdateChartResponse:
    """Update a chart asynchronously. See ``update()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("charts", "update_async")
    return await self._update_async(chart_id, request, project_id=project_id)

delete_async async

delete_async(
    chart_id: str, *, project_id: Optional[str] = None
) -> DeleteChartResponse

Delete a chart asynchronously. See delete().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def delete_async(
    self, chart_id: str, *, project_id: Optional[str] = None
) -> DeleteChartResponse:
    """Delete a chart asynchronously. See ``delete()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("charts", "delete_async")
    return await self._delete_async(chart_id, project_id=project_id)

query

query(
    request: ChartQueryRequest, *, project_id: str
) -> ChartQueryResponse

Aggregate a project's events into one chart result.

Accepted keys

Data plane fine-grained API key with project.event.get

Parameters:

Name	Type	Description	Default
request	ChartQueryRequest	The date range, filters, and query to run. The filters must include an event_type filter.	required
project_id	str	The project whose events are queried.	required

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

def query(
    self, request: ChartQueryRequest, *, project_id: str
) -> ChartQueryResponse:
    """Aggregate a project's events into one chart result.

    Accepted keys:
        Data plane fine-grained API key with ``project.event.get``

    Args:
        request: The date range, filters, and query to run. The filters
            must include an ``event_type`` filter.
        project_id: The project whose events are queried.
    """
    return charts_svc.queryChart(
        self._data_plane_api_config, project_id=project_id, data=request
    )

query_async async

query_async(
    request: ChartQueryRequest, *, project_id: str
) -> ChartQueryResponse

Aggregate a project's events asynchronously. See query().

Accepted keys

Data plane fine-grained API key with project.event.get

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

async def query_async(
    self, request: ChartQueryRequest, *, project_id: str
) -> ChartQueryResponse:
    """Aggregate a project's events asynchronously. See ``query()``.

    Accepted keys:
        Data plane fine-grained API key with ``project.event.get``
    """
    return await charts_svc_async.queryChart(
        self._data_plane_api_config, project_id=project_id, data=request
    )

ConfigurationsAPI

Bases: BaseAPI

Configurations API.

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

class ConfigurationsAPI(BaseAPI):
    """Configurations API."""

    # Sync methods
    def list(self, project: Optional[str] = None) -> List[ConfigurationItem]:
        """List configurations.

        Accepted keys:
            Project API key

        Note:
            The v1 API does not currently support project filtering.
        """
        if project is not None:
            warnings.warn(
                "The 'project' parameter is no longer supported for "
                "configurations.list() and will be removed in v2.0.",
                DeprecationWarning,
                stacklevel=2,
            )
        del project
        # Preserve the high-level SDK list surface by unwrapping the transport envelope.
        response = configs_svc.getConfigurations(self._api_config)
        return response.configurations

    def create(
        self, request: CreateConfigurationRequest
    ) -> CreateConfigurationResponse:
        """Create a configuration.

        Accepted keys:
            Project API key
        """
        return configs_svc.createConfiguration(self._api_config, data=request)

    def update(
        self, id: str, request: UpdateConfigurationRequest
    ) -> UpdateConfigurationResponse:
        """Update a configuration.

        Accepted keys:
            Project API key
        """
        return configs_svc.updateConfiguration(
            self._api_config, configId=id, data=request
        )

    def delete(self, id: str) -> DeleteConfigurationResponse:
        """Delete a configuration.

        Accepted keys:
            Project API key
        """
        return configs_svc.deleteConfiguration(self._api_config, configId=id)

    # Async methods
    async def list_async(
        self, project: Optional[str] = None
    ) -> List[ConfigurationItem]:
        """List configurations asynchronously.

        Accepted keys:
            Project API key

        Note:
            The v1 API does not currently support project filtering.
        """
        if project is not None:
            warnings.warn(
                "The 'project' parameter is no longer supported for "
                "configurations.list() and will be removed in v2.0.",
                DeprecationWarning,
                stacklevel=2,
            )
        del project
        response = await configs_svc_async.getConfigurations(self._api_config)
        return response.configurations

    async def create_async(
        self, request: CreateConfigurationRequest
    ) -> CreateConfigurationResponse:
        """Create a configuration asynchronously.

        Accepted keys:
            Project API key
        """
        return await configs_svc_async.createConfiguration(
            self._api_config, data=request
        )

    async def update_async(
        self, id: str, request: UpdateConfigurationRequest
    ) -> UpdateConfigurationResponse:
        """Update a configuration asynchronously.

        Accepted keys:
            Project API key
        """
        return await configs_svc_async.updateConfiguration(
            self._api_config, configId=id, data=request
        )

    async def delete_async(self, id: str) -> DeleteConfigurationResponse:
        """Delete a configuration asynchronously.

        Accepted keys:
            Project API key
        """
        return await configs_svc_async.deleteConfiguration(
            self._api_config, configId=id
        )

    # Backwards compatible aliases
    def get_configuration(self, id: str) -> List[ConfigurationItem]:
        """Get a configuration (backwards compatible alias).

        Accepted keys:
            Project API key
        """
        del id
        return self.list()  # No single-get endpoint, returns all

    def create_configuration(
        self, request: CreateConfigurationRequest
    ) -> CreateConfigurationResponse:
        """Create a configuration (backwards compatible alias).

        Accepted keys:
            Project API key
        """
        return self.create(request)

    def update_configuration(
        self, id: str, request: UpdateConfigurationRequest
    ) -> UpdateConfigurationResponse:
        """Update a configuration (backwards compatible alias).

        Accepted keys:
            Project API key
        """
        return self.update(id, request)

    def delete_configuration(self, id: str) -> DeleteConfigurationResponse:
        """Delete a configuration (backwards compatible alias).

        Accepted keys:
            Project API key
        """
        return self.delete(id)

    def list_configurations(
        self, project: Optional[str] = None
    ) -> List[ConfigurationItem]:
        """List configurations (backwards compatible alias).

        Accepted keys:
            Project API key
        """
        return self.list(project)

list

list(
    project: Optional[str] = None,
) -> List[ConfigurationItem]

List configurations.

Accepted keys

Project API key

Note

The v1 API does not currently support project filtering.

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

def list(self, project: Optional[str] = None) -> List[ConfigurationItem]:
    """List configurations.

    Accepted keys:
        Project API key

    Note:
        The v1 API does not currently support project filtering.
    """
    if project is not None:
        warnings.warn(
            "The 'project' parameter is no longer supported for "
            "configurations.list() and will be removed in v2.0.",
            DeprecationWarning,
            stacklevel=2,
        )
    del project
    # Preserve the high-level SDK list surface by unwrapping the transport envelope.
    response = configs_svc.getConfigurations(self._api_config)
    return response.configurations

create

create(
    request: CreateConfigurationRequest,
) -> CreateConfigurationResponse

Create a configuration.

Accepted keys

Project API key

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

def create(
    self, request: CreateConfigurationRequest
) -> CreateConfigurationResponse:
    """Create a configuration.

    Accepted keys:
        Project API key
    """
    return configs_svc.createConfiguration(self._api_config, data=request)

update

update(
    id: str, request: UpdateConfigurationRequest
) -> UpdateConfigurationResponse

Update a configuration.

Accepted keys

Project API key

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

def update(
    self, id: str, request: UpdateConfigurationRequest
) -> UpdateConfigurationResponse:
    """Update a configuration.

    Accepted keys:
        Project API key
    """
    return configs_svc.updateConfiguration(
        self._api_config, configId=id, data=request
    )

delete

delete(id: str) -> DeleteConfigurationResponse

Delete a configuration.

Accepted keys

Project API key

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

def delete(self, id: str) -> DeleteConfigurationResponse:
    """Delete a configuration.

    Accepted keys:
        Project API key
    """
    return configs_svc.deleteConfiguration(self._api_config, configId=id)

list_async async

list_async(
    project: Optional[str] = None,
) -> List[ConfigurationItem]

List configurations asynchronously.

Accepted keys

Project API key

Note

The v1 API does not currently support project filtering.

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

async def list_async(
    self, project: Optional[str] = None
) -> List[ConfigurationItem]:
    """List configurations asynchronously.

    Accepted keys:
        Project API key

    Note:
        The v1 API does not currently support project filtering.
    """
    if project is not None:
        warnings.warn(
            "The 'project' parameter is no longer supported for "
            "configurations.list() and will be removed in v2.0.",
            DeprecationWarning,
            stacklevel=2,
        )
    del project
    response = await configs_svc_async.getConfigurations(self._api_config)
    return response.configurations

create_async async

create_async(
    request: CreateConfigurationRequest,
) -> CreateConfigurationResponse

Create a configuration asynchronously.

Accepted keys

Project API key

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

async def create_async(
    self, request: CreateConfigurationRequest
) -> CreateConfigurationResponse:
    """Create a configuration asynchronously.

    Accepted keys:
        Project API key
    """
    return await configs_svc_async.createConfiguration(
        self._api_config, data=request
    )

update_async async

update_async(
    id: str, request: UpdateConfigurationRequest
) -> UpdateConfigurationResponse

Update a configuration asynchronously.

Accepted keys

Project API key

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

async def update_async(
    self, id: str, request: UpdateConfigurationRequest
) -> UpdateConfigurationResponse:
    """Update a configuration asynchronously.

    Accepted keys:
        Project API key
    """
    return await configs_svc_async.updateConfiguration(
        self._api_config, configId=id, data=request
    )

delete_async async

delete_async(id: str) -> DeleteConfigurationResponse

Delete a configuration asynchronously.

Accepted keys

Project API key

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

async def delete_async(self, id: str) -> DeleteConfigurationResponse:
    """Delete a configuration asynchronously.

    Accepted keys:
        Project API key
    """
    return await configs_svc_async.deleteConfiguration(
        self._api_config, configId=id
    )

get_configuration

get_configuration(id: str) -> List[ConfigurationItem]

Get a configuration (backwards compatible alias).

Accepted keys

Project API key

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

def get_configuration(self, id: str) -> List[ConfigurationItem]:
    """Get a configuration (backwards compatible alias).

    Accepted keys:
        Project API key
    """
    del id
    return self.list()  # No single-get endpoint, returns all

create_configuration

create_configuration(
    request: CreateConfigurationRequest,
) -> CreateConfigurationResponse

Create a configuration (backwards compatible alias).

Accepted keys

Project API key

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

def create_configuration(
    self, request: CreateConfigurationRequest
) -> CreateConfigurationResponse:
    """Create a configuration (backwards compatible alias).

    Accepted keys:
        Project API key
    """
    return self.create(request)

update_configuration

update_configuration(
    id: str, request: UpdateConfigurationRequest
) -> UpdateConfigurationResponse

Update a configuration (backwards compatible alias).

Accepted keys

Project API key

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

def update_configuration(
    self, id: str, request: UpdateConfigurationRequest
) -> UpdateConfigurationResponse:
    """Update a configuration (backwards compatible alias).

    Accepted keys:
        Project API key
    """
    return self.update(id, request)

delete_configuration

delete_configuration(
    id: str,
) -> DeleteConfigurationResponse

Delete a configuration (backwards compatible alias).

Accepted keys

Project API key

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

def delete_configuration(self, id: str) -> DeleteConfigurationResponse:
    """Delete a configuration (backwards compatible alias).

    Accepted keys:
        Project API key
    """
    return self.delete(id)

list_configurations

list_configurations(
    project: Optional[str] = None,
) -> List[ConfigurationItem]

List configurations (backwards compatible alias).

Accepted keys

Project API key

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

def list_configurations(
    self, project: Optional[str] = None
) -> List[ConfigurationItem]:
    """List configurations (backwards compatible alias).

    Accepted keys:
        Project API key
    """
    return self.list(project)

DatapointsAPI

Bases: BaseAPI

Datapoints API.

A method that takes project_id sends the request with the data plane API key when given it, naming the project in the path. Without it, the request acts on the project API key's project, is sent with that key, and issues a DeprecationWarning.

Each such public method, alias included, warns in its own name and then calls its private core, which sends the request and never warns. SDK code that reaches the operation calls the core, so the warning reaches only a caller who can pass project_id.

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

class DatapointsAPI(BaseAPI):
    """Datapoints API.

    A method that takes ``project_id`` sends the request with the data plane
    API key when given it, naming the project in the path. Without it, the
    request acts on the project API key's project, is sent with that key, and
    issues a ``DeprecationWarning``.

    Each such public method, alias included, warns in its own name and then
    calls its private core, which sends the request and never warns. SDK code
    that reaches the operation calls the core, so the warning reaches only a
    caller who can pass ``project_id``.
    """

    # Sync methods
    def list(
        self,
        datapoint_ids: Optional[List[str]] = None,
        dataset_name: Optional[str] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetDatapointsResponse:
        """List datapoints.

        When datapoint_ids exceeds QUERY_BATCH_SIZE, requests are automatically
        batched to stay within URL length limits and the results are merged.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            datapoint_ids: Optional list of datapoint IDs to fetch.
            dataset_name: Optional dataset name to filter by.
            project_id: The project whose datapoints are listed. Omitting it
                lists the project API key's datapoints, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("datapoints", "list")
        return self._list(datapoint_ids, dataset_name, project_id=project_id)

    def get(self, id: str, *, project_id: Optional[str] = None) -> GetDatapointResponse:
        """Get a datapoint by ID.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            id: The datapoint to get.
            project_id: The project that owns the datapoint. Omitting it reads
                from the project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("datapoints", "get")
        return self._get(id, project_id=project_id)

    def create(
        self, request: CreateDatapointRequest, *, project_id: Optional[str] = None
    ) -> CreateDatapointResponse:
        """Create a datapoint.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datapoints", "create")
        return self._create(request, project_id=project_id)

    def create_batch(
        self, request: BatchCreateDatapointsRequest, *, project_id: Optional[str] = None
    ) -> BatchCreateDatapointsResponse:
        """Create datapoints from the events ``request.checkState`` checks.

        A data plane API key needs both ``project.datapoint.post`` and
        ``project.event.get`` for the project, and also
        ``project.dataset.post`` when ``request.dataset_id`` is set.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datapoints", "create_batch")
        return self._create_batch(request, project_id=project_id)

    def update(
        self,
        id: str,
        request: UpdateDatapointRequest,
        *,
        project_id: Optional[str] = None,
    ) -> UpdateDatapointResponse:
        """Update a datapoint.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datapoints", "update")
        return self._update(id, request, project_id=project_id)

    def delete(
        self, id: str, *, project_id: Optional[str] = None
    ) -> DeleteDatapointResponse:
        """Delete a datapoint and remove it from every dataset that holds it.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            id: The datapoint to delete.
            project_id: The project that owns the datapoint. Omitting it deletes
                it from the project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("datapoints", "delete")
        return self._delete(id, project_id=project_id)

    # Async methods
    async def list_async(
        self,
        datapoint_ids: Optional[List[str]] = None,
        dataset_name: Optional[str] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetDatapointsResponse:
        """List datapoints asynchronously. See ``list()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datapoints", "list_async")
        return await self._list_async(
            datapoint_ids, dataset_name, project_id=project_id
        )

    async def get_async(
        self, id: str, *, project_id: Optional[str] = None
    ) -> GetDatapointResponse:
        """Get a datapoint by ID asynchronously. See ``get()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datapoints", "get_async")
        return await self._get_async(id, project_id=project_id)

    async def create_async(
        self, request: CreateDatapointRequest, *, project_id: Optional[str] = None
    ) -> CreateDatapointResponse:
        """Create a datapoint asynchronously.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datapoints", "create_async")
        return await self._create_async(request, project_id=project_id)

    async def create_batch_async(
        self, request: BatchCreateDatapointsRequest, *, project_id: Optional[str] = None
    ) -> BatchCreateDatapointsResponse:
        """Create datapoints from events asynchronously. See ``create_batch()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datapoints", "create_batch_async")
        return await self._create_batch_async(request, project_id=project_id)

    async def update_async(
        self,
        id: str,
        request: UpdateDatapointRequest,
        *,
        project_id: Optional[str] = None,
    ) -> UpdateDatapointResponse:
        """Update a datapoint asynchronously.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datapoints", "update_async")
        return await self._update_async(id, request, project_id=project_id)

    async def delete_async(
        self, id: str, *, project_id: Optional[str] = None
    ) -> DeleteDatapointResponse:
        """Delete a datapoint asynchronously. See ``delete()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datapoints", "delete_async")
        return await self._delete_async(id, project_id=project_id)

    # Backwards compatible aliases
    def get_datapoint(
        self, id: str, *, project_id: Optional[str] = None
    ) -> GetDatapointResponse:
        """Get a datapoint by ID (backwards compatible alias for get()).

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datapoints", "get_datapoint")
        return self._get(id, project_id=project_id)

    def create_datapoint(
        self, request: CreateDatapointRequest, *, project_id: Optional[str] = None
    ) -> CreateDatapointResponse:
        """Create a datapoint (backwards compatible alias for create()).

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datapoints", "create_datapoint")
        return self._create(request, project_id=project_id)

    def update_datapoint(
        self,
        id: str,
        request: UpdateDatapointRequest,
        *,
        project_id: Optional[str] = None,
    ) -> UpdateDatapointResponse:
        """Update a datapoint (backwards compatible alias for update()).

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datapoints", "update_datapoint")
        return self._update(id, request, project_id=project_id)

    def delete_datapoint(
        self, id: str, *, project_id: Optional[str] = None
    ) -> DeleteDatapointResponse:
        """Delete a datapoint (backwards compatible alias for delete()).

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datapoints", "delete_datapoint")
        return self._delete(id, project_id=project_id)

    def list_datapoints(
        self,
        datapoint_ids: Optional[List[str]] = None,
        dataset_name: Optional[str] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetDatapointsResponse:
        """List datapoints (backwards compatible alias for list()).

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datapoints", "list_datapoints")
        return self._list(datapoint_ids, dataset_name, project_id=project_id)

    # ---- Cores: send the request, never warn ----

    def _list(
        self,
        datapoint_ids: Optional[List[str]] = None,
        dataset_name: Optional[str] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetDatapointsResponse:
        """The request ``list()`` sends, one per ``QUERY_BATCH_SIZE`` chunk of ids."""

        def send(ids: Optional[List[str]]) -> GetDatapointsResponse:
            if project_id is None:
                return datapoints_svc.getDatapointsLegacy(
                    self._api_config, datapoint_ids=ids, dataset_name=dataset_name
                )
            return datapoints_svc.getDatapoints(
                self._data_plane_api_config,
                project_id=project_id,
                datapoint_ids=ids,
                dataset_name=dataset_name,
            )

        # Batch if the list is large enough to risk exceeding URL length limits.
        if datapoint_ids and len(datapoint_ids) > QUERY_BATCH_SIZE:
            all_datapoints: List[Any] = []
            batches = _chunk_list(datapoint_ids, QUERY_BATCH_SIZE)
            for i, batch in enumerate(batches):
                try:
                    resp = send(batch)
                except Exception:
                    logger.warning(
                        "Batch %d/%d failed (%d IDs)", i + 1, len(batches), len(batch)
                    )
                    raise
                all_datapoints.extend(resp.datapoints)
            # Skip re-validation — items are already validated Datapoint instances.
            return GetDatapointsResponse.model_construct(datapoints=all_datapoints)

        return send(datapoint_ids)

    async def _list_async(
        self,
        datapoint_ids: Optional[List[str]] = None,
        dataset_name: Optional[str] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetDatapointsResponse:
        """The request ``list_async()`` sends, one per ``QUERY_BATCH_SIZE`` chunk of ids."""

        async def send(ids: Optional[List[str]]) -> GetDatapointsResponse:
            if project_id is None:
                return await datapoints_svc_async.getDatapointsLegacy(
                    self._api_config, datapoint_ids=ids, dataset_name=dataset_name
                )
            return await datapoints_svc_async.getDatapoints(
                self._data_plane_api_config,
                project_id=project_id,
                datapoint_ids=ids,
                dataset_name=dataset_name,
            )

        # Batch if the list is large enough to risk exceeding URL length limits.
        if datapoint_ids and len(datapoint_ids) > QUERY_BATCH_SIZE:
            batches = _chunk_list(datapoint_ids, QUERY_BATCH_SIZE)
            resps = await asyncio.gather(*(send(batch) for batch in batches))
            all_datapoints: List[Any] = []
            for resp in resps:
                all_datapoints.extend(resp.datapoints)
            # Skip re-validation — items are already validated Datapoint instances.
            return GetDatapointsResponse.model_construct(datapoints=all_datapoints)

        return await send(datapoint_ids)

    def _get(
        self, id: str, *, project_id: Optional[str] = None
    ) -> GetDatapointResponse:
        """The request ``get()`` sends."""
        if project_id is None:
            return datapoints_svc.getDatapointLegacy(self._api_config, datapoint_id=id)
        return datapoints_svc.getDatapoint(
            self._data_plane_api_config, project_id=project_id, datapoint_id=id
        )

    async def _get_async(
        self, id: str, *, project_id: Optional[str] = None
    ) -> GetDatapointResponse:
        """The request ``get_async()`` sends."""
        if project_id is None:
            return await datapoints_svc_async.getDatapointLegacy(
                self._api_config, datapoint_id=id
            )
        return await datapoints_svc_async.getDatapoint(
            self._data_plane_api_config, project_id=project_id, datapoint_id=id
        )

    def _create(
        self, request: CreateDatapointRequest, *, project_id: Optional[str] = None
    ) -> CreateDatapointResponse:
        """The request ``create()`` sends."""
        if project_id is None:
            return datapoints_svc.createDatapointLegacy(self._api_config, data=request)
        return datapoints_svc.createDatapoint(
            self._data_plane_api_config, project_id=project_id, data=request
        )

    async def _create_async(
        self, request: CreateDatapointRequest, *, project_id: Optional[str] = None
    ) -> CreateDatapointResponse:
        """The request ``create_async()`` sends."""
        if project_id is None:
            return await datapoints_svc_async.createDatapointLegacy(
                self._api_config, data=request
            )
        return await datapoints_svc_async.createDatapoint(
            self._data_plane_api_config, project_id=project_id, data=request
        )

    def _create_batch(
        self, request: BatchCreateDatapointsRequest, *, project_id: Optional[str] = None
    ) -> BatchCreateDatapointsResponse:
        """The request ``create_batch()`` sends."""
        if project_id is None:
            return datapoints_svc.batchCreateDatapointsLegacy(
                self._api_config, data=request
            )
        return datapoints_svc.batchCreateDatapoints(
            self._data_plane_api_config, project_id=project_id, data=request
        )

    async def _create_batch_async(
        self, request: BatchCreateDatapointsRequest, *, project_id: Optional[str] = None
    ) -> BatchCreateDatapointsResponse:
        """The request ``create_batch_async()`` sends."""
        if project_id is None:
            return await datapoints_svc_async.batchCreateDatapointsLegacy(
                self._api_config, data=request
            )
        return await datapoints_svc_async.batchCreateDatapoints(
            self._data_plane_api_config, project_id=project_id, data=request
        )

    def _update(
        self,
        id: str,
        request: UpdateDatapointRequest,
        *,
        project_id: Optional[str] = None,
    ) -> UpdateDatapointResponse:
        """The request ``update()`` sends."""
        if project_id is None:
            return datapoints_svc.updateDatapointLegacy(
                self._api_config, datapoint_id=id, data=request
            )
        return datapoints_svc.updateDatapoint(
            self._data_plane_api_config,
            project_id=project_id,
            datapoint_id=id,
            data=request,
        )

    async def _update_async(
        self,
        id: str,
        request: UpdateDatapointRequest,
        *,
        project_id: Optional[str] = None,
    ) -> UpdateDatapointResponse:
        """The request ``update_async()`` sends."""
        if project_id is None:
            return await datapoints_svc_async.updateDatapointLegacy(
                self._api_config, datapoint_id=id, data=request
            )
        return await datapoints_svc_async.updateDatapoint(
            self._data_plane_api_config,
            project_id=project_id,
            datapoint_id=id,
            data=request,
        )

    def _delete(
        self, id: str, *, project_id: Optional[str] = None
    ) -> DeleteDatapointResponse:
        """The request ``delete()`` sends."""
        if project_id is None:
            return datapoints_svc.deleteDatapointLegacy(
                self._api_config, datapoint_id=id
            )
        return datapoints_svc.deleteDatapoint(
            self._data_plane_api_config, project_id=project_id, datapoint_id=id
        )

    async def _delete_async(
        self, id: str, *, project_id: Optional[str] = None
    ) -> DeleteDatapointResponse:
        """The request ``delete_async()`` sends."""
        if project_id is None:
            return await datapoints_svc_async.deleteDatapointLegacy(
                self._api_config, datapoint_id=id
            )
        return await datapoints_svc_async.deleteDatapoint(
            self._data_plane_api_config, project_id=project_id, datapoint_id=id
        )

list

list(
    datapoint_ids: Optional[List[str]] = None,
    dataset_name: Optional[str] = None,
    *,
    project_id: Optional[str] = None
) -> GetDatapointsResponse

List datapoints.

When datapoint_ids exceeds QUERY_BATCH_SIZE, requests are automatically batched to stay within URL length limits and the results are merged.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
datapoint_ids	Optional[List[str]]	Optional list of datapoint IDs to fetch.	None
dataset_name	Optional[str]	Optional dataset name to filter by.	None
project_id	Optional[str]	The project whose datapoints are listed. Omitting it lists the project API key's datapoints, which is deprecated.	None

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

def list(
    self,
    datapoint_ids: Optional[List[str]] = None,
    dataset_name: Optional[str] = None,
    *,
    project_id: Optional[str] = None,
) -> GetDatapointsResponse:
    """List datapoints.

    When datapoint_ids exceeds QUERY_BATCH_SIZE, requests are automatically
    batched to stay within URL length limits and the results are merged.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        datapoint_ids: Optional list of datapoint IDs to fetch.
        dataset_name: Optional dataset name to filter by.
        project_id: The project whose datapoints are listed. Omitting it
            lists the project API key's datapoints, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("datapoints", "list")
    return self._list(datapoint_ids, dataset_name, project_id=project_id)

get

get(
    id: str, *, project_id: Optional[str] = None
) -> GetDatapointResponse

Get a datapoint by ID.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
id	str	The datapoint to get.	required
project_id	Optional[str]	The project that owns the datapoint. Omitting it reads from the project API key's project, which is deprecated.	None

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

def get(self, id: str, *, project_id: Optional[str] = None) -> GetDatapointResponse:
    """Get a datapoint by ID.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        id: The datapoint to get.
        project_id: The project that owns the datapoint. Omitting it reads
            from the project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("datapoints", "get")
    return self._get(id, project_id=project_id)

create

create(
    request: CreateDatapointRequest,
    *,
    project_id: Optional[str] = None
) -> CreateDatapointResponse

Create a datapoint.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

def create(
    self, request: CreateDatapointRequest, *, project_id: Optional[str] = None
) -> CreateDatapointResponse:
    """Create a datapoint.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datapoints", "create")
    return self._create(request, project_id=project_id)

create_batch

create_batch(
    request: BatchCreateDatapointsRequest,
    *,
    project_id: Optional[str] = None
) -> BatchCreateDatapointsResponse

Create datapoints from the events request.checkState checks.

A data plane API key needs both project.datapoint.post and project.event.get for the project, and also project.dataset.post when request.dataset_id is set.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

def create_batch(
    self, request: BatchCreateDatapointsRequest, *, project_id: Optional[str] = None
) -> BatchCreateDatapointsResponse:
    """Create datapoints from the events ``request.checkState`` checks.

    A data plane API key needs both ``project.datapoint.post`` and
    ``project.event.get`` for the project, and also
    ``project.dataset.post`` when ``request.dataset_id`` is set.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datapoints", "create_batch")
    return self._create_batch(request, project_id=project_id)

update

update(
    id: str,
    request: UpdateDatapointRequest,
    *,
    project_id: Optional[str] = None
) -> UpdateDatapointResponse

Update a datapoint.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

def update(
    self,
    id: str,
    request: UpdateDatapointRequest,
    *,
    project_id: Optional[str] = None,
) -> UpdateDatapointResponse:
    """Update a datapoint.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datapoints", "update")
    return self._update(id, request, project_id=project_id)

delete

delete(
    id: str, *, project_id: Optional[str] = None
) -> DeleteDatapointResponse

Delete a datapoint and remove it from every dataset that holds it.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
id	str	The datapoint to delete.	required
project_id	Optional[str]	The project that owns the datapoint. Omitting it deletes it from the project API key's project, which is deprecated.	None

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

def delete(
    self, id: str, *, project_id: Optional[str] = None
) -> DeleteDatapointResponse:
    """Delete a datapoint and remove it from every dataset that holds it.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        id: The datapoint to delete.
        project_id: The project that owns the datapoint. Omitting it deletes
            it from the project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("datapoints", "delete")
    return self._delete(id, project_id=project_id)

list_async async

list_async(
    datapoint_ids: Optional[List[str]] = None,
    dataset_name: Optional[str] = None,
    *,
    project_id: Optional[str] = None
) -> GetDatapointsResponse

List datapoints asynchronously. See list().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def list_async(
    self,
    datapoint_ids: Optional[List[str]] = None,
    dataset_name: Optional[str] = None,
    *,
    project_id: Optional[str] = None,
) -> GetDatapointsResponse:
    """List datapoints asynchronously. See ``list()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datapoints", "list_async")
    return await self._list_async(
        datapoint_ids, dataset_name, project_id=project_id
    )

get_async async

get_async(
    id: str, *, project_id: Optional[str] = None
) -> GetDatapointResponse

Get a datapoint by ID asynchronously. See get().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def get_async(
    self, id: str, *, project_id: Optional[str] = None
) -> GetDatapointResponse:
    """Get a datapoint by ID asynchronously. See ``get()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datapoints", "get_async")
    return await self._get_async(id, project_id=project_id)

create_async async

create_async(
    request: CreateDatapointRequest,
    *,
    project_id: Optional[str] = None
) -> CreateDatapointResponse

Create a datapoint asynchronously.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def create_async(
    self, request: CreateDatapointRequest, *, project_id: Optional[str] = None
) -> CreateDatapointResponse:
    """Create a datapoint asynchronously.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datapoints", "create_async")
    return await self._create_async(request, project_id=project_id)

create_batch_async async

create_batch_async(
    request: BatchCreateDatapointsRequest,
    *,
    project_id: Optional[str] = None
) -> BatchCreateDatapointsResponse

Create datapoints from events asynchronously. See create_batch().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def create_batch_async(
    self, request: BatchCreateDatapointsRequest, *, project_id: Optional[str] = None
) -> BatchCreateDatapointsResponse:
    """Create datapoints from events asynchronously. See ``create_batch()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datapoints", "create_batch_async")
    return await self._create_batch_async(request, project_id=project_id)

update_async async

update_async(
    id: str,
    request: UpdateDatapointRequest,
    *,
    project_id: Optional[str] = None
) -> UpdateDatapointResponse

Update a datapoint asynchronously.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def update_async(
    self,
    id: str,
    request: UpdateDatapointRequest,
    *,
    project_id: Optional[str] = None,
) -> UpdateDatapointResponse:
    """Update a datapoint asynchronously.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datapoints", "update_async")
    return await self._update_async(id, request, project_id=project_id)

delete_async async

delete_async(
    id: str, *, project_id: Optional[str] = None
) -> DeleteDatapointResponse

Delete a datapoint asynchronously. See delete().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def delete_async(
    self, id: str, *, project_id: Optional[str] = None
) -> DeleteDatapointResponse:
    """Delete a datapoint asynchronously. See ``delete()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datapoints", "delete_async")
    return await self._delete_async(id, project_id=project_id)

get_datapoint

get_datapoint(
    id: str, *, project_id: Optional[str] = None
) -> GetDatapointResponse

Get a datapoint by ID (backwards compatible alias for get()).

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

def get_datapoint(
    self, id: str, *, project_id: Optional[str] = None
) -> GetDatapointResponse:
    """Get a datapoint by ID (backwards compatible alias for get()).

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datapoints", "get_datapoint")
    return self._get(id, project_id=project_id)

create_datapoint

create_datapoint(
    request: CreateDatapointRequest,
    *,
    project_id: Optional[str] = None
) -> CreateDatapointResponse

Create a datapoint (backwards compatible alias for create()).

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

def create_datapoint(
    self, request: CreateDatapointRequest, *, project_id: Optional[str] = None
) -> CreateDatapointResponse:
    """Create a datapoint (backwards compatible alias for create()).

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datapoints", "create_datapoint")
    return self._create(request, project_id=project_id)

update_datapoint

update_datapoint(
    id: str,
    request: UpdateDatapointRequest,
    *,
    project_id: Optional[str] = None
) -> UpdateDatapointResponse

Update a datapoint (backwards compatible alias for update()).

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

def update_datapoint(
    self,
    id: str,
    request: UpdateDatapointRequest,
    *,
    project_id: Optional[str] = None,
) -> UpdateDatapointResponse:
    """Update a datapoint (backwards compatible alias for update()).

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datapoints", "update_datapoint")
    return self._update(id, request, project_id=project_id)

delete_datapoint

delete_datapoint(
    id: str, *, project_id: Optional[str] = None
) -> DeleteDatapointResponse

Delete a datapoint (backwards compatible alias for delete()).

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

def delete_datapoint(
    self, id: str, *, project_id: Optional[str] = None
) -> DeleteDatapointResponse:
    """Delete a datapoint (backwards compatible alias for delete()).

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datapoints", "delete_datapoint")
    return self._delete(id, project_id=project_id)

list_datapoints

list_datapoints(
    datapoint_ids: Optional[List[str]] = None,
    dataset_name: Optional[str] = None,
    *,
    project_id: Optional[str] = None
) -> GetDatapointsResponse

List datapoints (backwards compatible alias for list()).

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

def list_datapoints(
    self,
    datapoint_ids: Optional[List[str]] = None,
    dataset_name: Optional[str] = None,
    *,
    project_id: Optional[str] = None,
) -> GetDatapointsResponse:
    """List datapoints (backwards compatible alias for list()).

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datapoints", "list_datapoints")
    return self._list(datapoint_ids, dataset_name, project_id=project_id)

DatasetsAPI

Bases: BaseAPI

Datasets API.

A method that takes project_id sends the request with the data plane API key when given it, naming the project in the path. Without it, the request acts on the project API key's project, is sent with that key, and issues a DeprecationWarning.

Each such public method, alias included, warns in its own name and then calls its private core, which sends the request and never warns. SDK code that reaches the operation calls the core, so the warning reaches only a caller who can pass project_id.

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

class DatasetsAPI(BaseAPI):
    """Datasets API.

    A method that takes ``project_id`` sends the request with the data plane
    API key when given it, naming the project in the path. Without it, the
    request acts on the project API key's project, is sent with that key, and
    issues a ``DeprecationWarning``.

    Each such public method, alias included, warns in its own name and then
    calls its private core, which sends the request and never warns. SDK code
    that reaches the operation calls the core, so the warning reaches only a
    caller who can pass ``project_id``.
    """

    # Sync methods
    def list(
        self,
        dataset_id: Optional[str] = None,
        name: Optional[str] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetDatasetsResponse:
        """List datasets.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            dataset_id: Optional dataset ID to fetch.
            name: Optional dataset name to filter by.
            project_id: The project whose datasets are listed. Omitting it lists
                the project API key's datasets, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("datasets", "list")
        return self._list(dataset_id, name, project_id=project_id)

    def create(
        self, request: CreateDatasetRequest, *, project_id: Optional[str] = None
    ) -> CreateDatasetResponse:
        """Create a dataset.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datasets", "create")
        return self._create(request, project_id=project_id)

    def update(
        self, request: UpdateDatasetRequest, *, project_id: Optional[str] = None
    ) -> UpdateDatasetResponse:
        """Update a dataset.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            request: The fields to update, with the dataset's ``dataset_id``.
            project_id: The project that owns the dataset. Omitting it updates
                it in the project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("datasets", "update")
        return self._update(request, project_id=project_id)

    def delete(
        self, id: str, *, project_id: Optional[str] = None
    ) -> DeleteDatasetResponse:
        """Delete a dataset.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            id: The dataset to delete.
            project_id: The project that owns the dataset. Omitting it deletes
                it from the project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("datasets", "delete")
        return self._delete(id, project_id=project_id)

    def add_datapoints(
        self,
        dataset_id: str,
        request: AddDatapointsToDatasetRequest,
        *,
        project_id: Optional[str] = None,
    ) -> AddDatapointsResponse:
        """Add datapoints to a dataset.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            dataset_id: The unique identifier of the dataset to add datapoints to.
            request: The request containing data and mapping for the datapoints.
            project_id: The project that owns the dataset. Omitting it adds to
                the dataset in the project API key's project, which is
                deprecated.

        Returns:
            AddDatapointsResponse with inserted status and datapoint IDs.
        """
        if project_id is None:
            _warn_call_without_project_id("datasets", "add_datapoints")
        return self._add_datapoints(dataset_id, request, project_id=project_id)

    def remove_datapoint(
        self,
        dataset_id: str,
        datapoint_id: str,
        *,
        project_id: Optional[str] = None,
    ) -> RemoveDatapointResponse:
        """Remove a datapoint from a dataset.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            dataset_id: The unique identifier of the dataset.
            datapoint_id: The unique identifier of the datapoint to remove.
            project_id: The project that owns the dataset. Omitting it removes
                the datapoint from the dataset in the project API key's
                project, which is deprecated.

        Returns:
            RemoveDatapointResponse with dereferenced status and message.
        """
        if project_id is None:
            _warn_call_without_project_id("datasets", "remove_datapoint")
        return self._remove_datapoint(dataset_id, datapoint_id, project_id=project_id)

    # Async methods
    async def list_async(
        self,
        dataset_id: Optional[str] = None,
        name: Optional[str] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetDatasetsResponse:
        """List datasets asynchronously. See ``list()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datasets", "list_async")
        return await self._list_async(dataset_id, name, project_id=project_id)

    async def create_async(
        self, request: CreateDatasetRequest, *, project_id: Optional[str] = None
    ) -> CreateDatasetResponse:
        """Create a dataset asynchronously.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datasets", "create_async")
        return await self._create_async(request, project_id=project_id)

    async def update_async(
        self, request: UpdateDatasetRequest, *, project_id: Optional[str] = None
    ) -> UpdateDatasetResponse:
        """Update a dataset asynchronously. See ``update()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datasets", "update_async")
        return await self._update_async(request, project_id=project_id)

    async def delete_async(
        self, id: str, *, project_id: Optional[str] = None
    ) -> DeleteDatasetResponse:
        """Delete a dataset asynchronously. See ``delete()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datasets", "delete_async")
        return await self._delete_async(id, project_id=project_id)

    async def add_datapoints_async(
        self,
        dataset_id: str,
        request: AddDatapointsToDatasetRequest,
        *,
        project_id: Optional[str] = None,
    ) -> AddDatapointsResponse:
        """Add datapoints to a dataset asynchronously. See ``add_datapoints()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datasets", "add_datapoints_async")
        return await self._add_datapoints_async(
            dataset_id, request, project_id=project_id
        )

    async def remove_datapoint_async(
        self,
        dataset_id: str,
        datapoint_id: str,
        *,
        project_id: Optional[str] = None,
    ) -> RemoveDatapointResponse:
        """Remove a datapoint from a dataset asynchronously. See ``remove_datapoint()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datasets", "remove_datapoint_async")
        return await self._remove_datapoint_async(
            dataset_id, datapoint_id, project_id=project_id
        )

    # Backwards compatible aliases
    def get_dataset(
        self, id: str, *, project_id: Optional[str] = None
    ) -> GetDatasetsResponse:
        """Get a dataset by ID (backwards compatible alias).

        Sends the list request filtered to ``id``, because there is no
        single-dataset operation.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datasets", "get_dataset")
        return self._list(dataset_id=id, project_id=project_id)

    def create_dataset(
        self, request: CreateDatasetRequest, *, project_id: Optional[str] = None
    ) -> CreateDatasetResponse:
        """Create a dataset (backwards compatible alias for create()).

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datasets", "create_dataset")
        return self._create(request, project_id=project_id)

    def update_dataset(
        self, request: UpdateDatasetRequest, *, project_id: Optional[str] = None
    ) -> UpdateDatasetResponse:
        """Update a dataset (backwards compatible alias for update()).

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datasets", "update_dataset")
        return self._update(request, project_id=project_id)

    def delete_dataset(
        self, id: str, *, project_id: Optional[str] = None
    ) -> DeleteDatasetResponse:
        """Delete a dataset (backwards compatible alias for delete()).

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datasets", "delete_dataset")
        return self._delete(id, project_id=project_id)

    def list_datasets(
        self,
        dataset_id: Optional[str] = None,
        name: Optional[str] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetDatasetsResponse:
        """List datasets (backwards compatible alias).

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("datasets", "list_datasets")
        return self._list(dataset_id, name, project_id=project_id)

    # ---- Cores: send the request, never warn ----

    def _list(
        self,
        dataset_id: Optional[str] = None,
        name: Optional[str] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetDatasetsResponse:
        """The request ``list()`` sends."""
        if project_id is None:
            return datasets_svc.getDatasetsLegacy(
                self._api_config, dataset_id=dataset_id, name=name
            )
        return datasets_svc.getDatasets(
            self._data_plane_api_config,
            project_id=project_id,
            dataset_id=dataset_id,
            name=name,
        )

    async def _list_async(
        self,
        dataset_id: Optional[str] = None,
        name: Optional[str] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetDatasetsResponse:
        """The request ``list_async()`` sends."""
        if project_id is None:
            return await datasets_svc_async.getDatasetsLegacy(
                self._api_config, dataset_id=dataset_id, name=name
            )
        return await datasets_svc_async.getDatasets(
            self._data_plane_api_config,
            project_id=project_id,
            dataset_id=dataset_id,
            name=name,
        )

    def _create(
        self, request: CreateDatasetRequest, *, project_id: Optional[str] = None
    ) -> CreateDatasetResponse:
        """The request ``create()`` sends."""
        if project_id is None:
            return datasets_svc.createDatasetLegacy(self._api_config, data=request)
        return datasets_svc.createDataset(
            self._data_plane_api_config, project_id=project_id, data=request
        )

    async def _create_async(
        self, request: CreateDatasetRequest, *, project_id: Optional[str] = None
    ) -> CreateDatasetResponse:
        """The request ``create_async()`` sends."""
        if project_id is None:
            return await datasets_svc_async.createDatasetLegacy(
                self._api_config, data=request
            )
        return await datasets_svc_async.createDataset(
            self._data_plane_api_config, project_id=project_id, data=request
        )

    def _update(
        self, request: UpdateDatasetRequest, *, project_id: Optional[str] = None
    ) -> UpdateDatasetResponse:
        """The request ``update()`` sends."""
        if project_id is None:
            return datasets_svc.updateDatasetBodyIdLegacy(
                self._api_config, data=request
            )
        dataset_id, body = _split_dataset_update(request)
        return datasets_svc.updateDataset(
            self._data_plane_api_config,
            project_id=project_id,
            dataset_id=dataset_id,
            data=body,
        )

    async def _update_async(
        self, request: UpdateDatasetRequest, *, project_id: Optional[str] = None
    ) -> UpdateDatasetResponse:
        """The request ``update_async()`` sends."""
        if project_id is None:
            return await datasets_svc_async.updateDatasetBodyIdLegacy(
                self._api_config, data=request
            )
        dataset_id, body = _split_dataset_update(request)
        return await datasets_svc_async.updateDataset(
            self._data_plane_api_config,
            project_id=project_id,
            dataset_id=dataset_id,
            data=body,
        )

    def _delete(
        self, id: str, *, project_id: Optional[str] = None
    ) -> DeleteDatasetResponse:
        """The request ``delete()`` sends."""
        if project_id is None:
            return datasets_svc.deleteDatasetQueryIdLegacy(
                self._api_config, dataset_id=id
            )
        return datasets_svc.deleteDataset(
            self._data_plane_api_config, project_id=project_id, dataset_id=id
        )

    async def _delete_async(
        self, id: str, *, project_id: Optional[str] = None
    ) -> DeleteDatasetResponse:
        """The request ``delete_async()`` sends."""
        if project_id is None:
            return await datasets_svc_async.deleteDatasetQueryIdLegacy(
                self._api_config, dataset_id=id
            )
        return await datasets_svc_async.deleteDataset(
            self._data_plane_api_config, project_id=project_id, dataset_id=id
        )

    def _add_datapoints(
        self,
        dataset_id: str,
        request: AddDatapointsToDatasetRequest,
        *,
        project_id: Optional[str] = None,
    ) -> AddDatapointsResponse:
        """The request ``add_datapoints()`` sends."""
        if project_id is None:
            return datasets_svc.addDatapointsLegacy(
                self._api_config, dataset_id=dataset_id, data=request
            )
        return datasets_svc.addDatapoints(
            self._data_plane_api_config,
            project_id=project_id,
            dataset_id=dataset_id,
            data=request,
        )

    async def _add_datapoints_async(
        self,
        dataset_id: str,
        request: AddDatapointsToDatasetRequest,
        *,
        project_id: Optional[str] = None,
    ) -> AddDatapointsResponse:
        """The request ``add_datapoints_async()`` sends."""
        if project_id is None:
            return await datasets_svc_async.addDatapointsLegacy(
                self._api_config, dataset_id=dataset_id, data=request
            )
        return await datasets_svc_async.addDatapoints(
            self._data_plane_api_config,
            project_id=project_id,
            dataset_id=dataset_id,
            data=request,
        )

    def _remove_datapoint(
        self,
        dataset_id: str,
        datapoint_id: str,
        *,
        project_id: Optional[str] = None,
    ) -> RemoveDatapointResponse:
        """The request ``remove_datapoint()`` sends."""
        if project_id is None:
            return datasets_svc.removeDatapointShortPathLegacy(
                self._api_config, dataset_id=dataset_id, datapoint_id=datapoint_id
            )
        return datasets_svc.removeDatapoint(
            self._data_plane_api_config,
            project_id=project_id,
            dataset_id=dataset_id,
            datapoint_id=datapoint_id,
        )

    async def _remove_datapoint_async(
        self,
        dataset_id: str,
        datapoint_id: str,
        *,
        project_id: Optional[str] = None,
    ) -> RemoveDatapointResponse:
        """The request ``remove_datapoint_async()`` sends."""
        if project_id is None:
            return await datasets_svc_async.removeDatapointShortPathLegacy(
                self._api_config, dataset_id=dataset_id, datapoint_id=datapoint_id
            )
        return await datasets_svc_async.removeDatapoint(
            self._data_plane_api_config,
            project_id=project_id,
            dataset_id=dataset_id,
            datapoint_id=datapoint_id,
        )

list

list(
    dataset_id: Optional[str] = None,
    name: Optional[str] = None,
    *,
    project_id: Optional[str] = None
) -> GetDatasetsResponse

List datasets.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
dataset_id	Optional[str]	Optional dataset ID to fetch.	None
name	Optional[str]	Optional dataset name to filter by.	None
project_id	Optional[str]	The project whose datasets are listed. Omitting it lists the project API key's datasets, which is deprecated.	None

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

def list(
    self,
    dataset_id: Optional[str] = None,
    name: Optional[str] = None,
    *,
    project_id: Optional[str] = None,
) -> GetDatasetsResponse:
    """List datasets.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        dataset_id: Optional dataset ID to fetch.
        name: Optional dataset name to filter by.
        project_id: The project whose datasets are listed. Omitting it lists
            the project API key's datasets, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("datasets", "list")
    return self._list(dataset_id, name, project_id=project_id)

create

create(
    request: CreateDatasetRequest,
    *,
    project_id: Optional[str] = None
) -> CreateDatasetResponse

Create a dataset.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

def create(
    self, request: CreateDatasetRequest, *, project_id: Optional[str] = None
) -> CreateDatasetResponse:
    """Create a dataset.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datasets", "create")
    return self._create(request, project_id=project_id)

update

update(
    request: UpdateDatasetRequest,
    *,
    project_id: Optional[str] = None
) -> UpdateDatasetResponse

Update a dataset.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
request	UpdateDatasetRequest	The fields to update, with the dataset's dataset_id.	required
project_id	Optional[str]	The project that owns the dataset. Omitting it updates it in the project API key's project, which is deprecated.	None

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

def update(
    self, request: UpdateDatasetRequest, *, project_id: Optional[str] = None
) -> UpdateDatasetResponse:
    """Update a dataset.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        request: The fields to update, with the dataset's ``dataset_id``.
        project_id: The project that owns the dataset. Omitting it updates
            it in the project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("datasets", "update")
    return self._update(request, project_id=project_id)

delete

delete(
    id: str, *, project_id: Optional[str] = None
) -> DeleteDatasetResponse

Delete a dataset.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
id	str	The dataset to delete.	required
project_id	Optional[str]	The project that owns the dataset. Omitting it deletes it from the project API key's project, which is deprecated.	None

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

def delete(
    self, id: str, *, project_id: Optional[str] = None
) -> DeleteDatasetResponse:
    """Delete a dataset.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        id: The dataset to delete.
        project_id: The project that owns the dataset. Omitting it deletes
            it from the project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("datasets", "delete")
    return self._delete(id, project_id=project_id)

add_datapoints

add_datapoints(
    dataset_id: str,
    request: AddDatapointsToDatasetRequest,
    *,
    project_id: Optional[str] = None
) -> AddDatapointsResponse

Add datapoints to a dataset.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
dataset_id	str	The unique identifier of the dataset to add datapoints to.	required
request	AddDatapointsToDatasetRequest	The request containing data and mapping for the datapoints.	required
project_id	Optional[str]	The project that owns the dataset. Omitting it adds to the dataset in the project API key's project, which is deprecated.	None

Returns:

Type	Description
AddDatapointsResponse	AddDatapointsResponse with inserted status and datapoint IDs.

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

def add_datapoints(
    self,
    dataset_id: str,
    request: AddDatapointsToDatasetRequest,
    *,
    project_id: Optional[str] = None,
) -> AddDatapointsResponse:
    """Add datapoints to a dataset.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        dataset_id: The unique identifier of the dataset to add datapoints to.
        request: The request containing data and mapping for the datapoints.
        project_id: The project that owns the dataset. Omitting it adds to
            the dataset in the project API key's project, which is
            deprecated.

    Returns:
        AddDatapointsResponse with inserted status and datapoint IDs.
    """
    if project_id is None:
        _warn_call_without_project_id("datasets", "add_datapoints")
    return self._add_datapoints(dataset_id, request, project_id=project_id)

remove_datapoint

remove_datapoint(
    dataset_id: str,
    datapoint_id: str,
    *,
    project_id: Optional[str] = None
) -> RemoveDatapointResponse

Remove a datapoint from a dataset.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
dataset_id	str	The unique identifier of the dataset.	required
datapoint_id	str	The unique identifier of the datapoint to remove.	required
project_id	Optional[str]	The project that owns the dataset. Omitting it removes the datapoint from the dataset in the project API key's project, which is deprecated.	None

Returns:

Type	Description
RemoveDatapointResponse	RemoveDatapointResponse with dereferenced status and message.

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

def remove_datapoint(
    self,
    dataset_id: str,
    datapoint_id: str,
    *,
    project_id: Optional[str] = None,
) -> RemoveDatapointResponse:
    """Remove a datapoint from a dataset.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        dataset_id: The unique identifier of the dataset.
        datapoint_id: The unique identifier of the datapoint to remove.
        project_id: The project that owns the dataset. Omitting it removes
            the datapoint from the dataset in the project API key's
            project, which is deprecated.

    Returns:
        RemoveDatapointResponse with dereferenced status and message.
    """
    if project_id is None:
        _warn_call_without_project_id("datasets", "remove_datapoint")
    return self._remove_datapoint(dataset_id, datapoint_id, project_id=project_id)

list_async async

list_async(
    dataset_id: Optional[str] = None,
    name: Optional[str] = None,
    *,
    project_id: Optional[str] = None
) -> GetDatasetsResponse

List datasets asynchronously. See list().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def list_async(
    self,
    dataset_id: Optional[str] = None,
    name: Optional[str] = None,
    *,
    project_id: Optional[str] = None,
) -> GetDatasetsResponse:
    """List datasets asynchronously. See ``list()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datasets", "list_async")
    return await self._list_async(dataset_id, name, project_id=project_id)

create_async async

create_async(
    request: CreateDatasetRequest,
    *,
    project_id: Optional[str] = None
) -> CreateDatasetResponse

Create a dataset asynchronously.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def create_async(
    self, request: CreateDatasetRequest, *, project_id: Optional[str] = None
) -> CreateDatasetResponse:
    """Create a dataset asynchronously.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datasets", "create_async")
    return await self._create_async(request, project_id=project_id)

update_async async

update_async(
    request: UpdateDatasetRequest,
    *,
    project_id: Optional[str] = None
) -> UpdateDatasetResponse

Update a dataset asynchronously. See update().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def update_async(
    self, request: UpdateDatasetRequest, *, project_id: Optional[str] = None
) -> UpdateDatasetResponse:
    """Update a dataset asynchronously. See ``update()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datasets", "update_async")
    return await self._update_async(request, project_id=project_id)

delete_async async

delete_async(
    id: str, *, project_id: Optional[str] = None
) -> DeleteDatasetResponse

Delete a dataset asynchronously. See delete().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def delete_async(
    self, id: str, *, project_id: Optional[str] = None
) -> DeleteDatasetResponse:
    """Delete a dataset asynchronously. See ``delete()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datasets", "delete_async")
    return await self._delete_async(id, project_id=project_id)

add_datapoints_async async

add_datapoints_async(
    dataset_id: str,
    request: AddDatapointsToDatasetRequest,
    *,
    project_id: Optional[str] = None
) -> AddDatapointsResponse

Add datapoints to a dataset asynchronously. See add_datapoints().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def add_datapoints_async(
    self,
    dataset_id: str,
    request: AddDatapointsToDatasetRequest,
    *,
    project_id: Optional[str] = None,
) -> AddDatapointsResponse:
    """Add datapoints to a dataset asynchronously. See ``add_datapoints()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datasets", "add_datapoints_async")
    return await self._add_datapoints_async(
        dataset_id, request, project_id=project_id
    )

remove_datapoint_async async

remove_datapoint_async(
    dataset_id: str,
    datapoint_id: str,
    *,
    project_id: Optional[str] = None
) -> RemoveDatapointResponse

Remove a datapoint from a dataset asynchronously. See remove_datapoint().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def remove_datapoint_async(
    self,
    dataset_id: str,
    datapoint_id: str,
    *,
    project_id: Optional[str] = None,
) -> RemoveDatapointResponse:
    """Remove a datapoint from a dataset asynchronously. See ``remove_datapoint()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datasets", "remove_datapoint_async")
    return await self._remove_datapoint_async(
        dataset_id, datapoint_id, project_id=project_id
    )

get_dataset

get_dataset(
    id: str, *, project_id: Optional[str] = None
) -> GetDatasetsResponse

Get a dataset by ID (backwards compatible alias).

Sends the list request filtered to id, because there is no single-dataset operation.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

def get_dataset(
    self, id: str, *, project_id: Optional[str] = None
) -> GetDatasetsResponse:
    """Get a dataset by ID (backwards compatible alias).

    Sends the list request filtered to ``id``, because there is no
    single-dataset operation.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datasets", "get_dataset")
    return self._list(dataset_id=id, project_id=project_id)

create_dataset

create_dataset(
    request: CreateDatasetRequest,
    *,
    project_id: Optional[str] = None
) -> CreateDatasetResponse

Create a dataset (backwards compatible alias for create()).

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

def create_dataset(
    self, request: CreateDatasetRequest, *, project_id: Optional[str] = None
) -> CreateDatasetResponse:
    """Create a dataset (backwards compatible alias for create()).

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datasets", "create_dataset")
    return self._create(request, project_id=project_id)

update_dataset

update_dataset(
    request: UpdateDatasetRequest,
    *,
    project_id: Optional[str] = None
) -> UpdateDatasetResponse

Update a dataset (backwards compatible alias for update()).

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

def update_dataset(
    self, request: UpdateDatasetRequest, *, project_id: Optional[str] = None
) -> UpdateDatasetResponse:
    """Update a dataset (backwards compatible alias for update()).

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datasets", "update_dataset")
    return self._update(request, project_id=project_id)

delete_dataset

delete_dataset(
    id: str, *, project_id: Optional[str] = None
) -> DeleteDatasetResponse

Delete a dataset (backwards compatible alias for delete()).

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

def delete_dataset(
    self, id: str, *, project_id: Optional[str] = None
) -> DeleteDatasetResponse:
    """Delete a dataset (backwards compatible alias for delete()).

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datasets", "delete_dataset")
    return self._delete(id, project_id=project_id)

list_datasets

list_datasets(
    dataset_id: Optional[str] = None,
    name: Optional[str] = None,
    *,
    project_id: Optional[str] = None
) -> GetDatasetsResponse

List datasets (backwards compatible alias).

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

def list_datasets(
    self,
    dataset_id: Optional[str] = None,
    name: Optional[str] = None,
    *,
    project_id: Optional[str] = None,
) -> GetDatasetsResponse:
    """List datasets (backwards compatible alias).

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("datasets", "list_datasets")
    return self._list(dataset_id, name, project_id=project_id)

EventsAPI

Bases: BaseAPI

Events API.

Event reads and writes now go through Data Plane endpoints. Legacy helpers like list() and get_by_session_id() are retained for backward compatibility on top of the export endpoint.

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

class EventsAPI(BaseAPI):
    """Events API.

    Event reads and writes now go through Data Plane endpoints. Legacy helpers
    like list() and get_by_session_id() are retained for backward compatibility
    on top of the export endpoint.
    """

    def __init__(
        self,
        api_config: APIConfig,
        ingestion_api_key: str = "",
        *,
        retry_config: Optional[RetryConfig] = None,
    ) -> None:
        super().__init__(api_config, ingestion_api_key)
        self._retry_config = RetryConfig.from_optional(retry_config)

    # Supported parameters for getEvents() method
    _GET_EVENTS_SUPPORTED_PARAMS = {
        "dateRange",
        "filters",
        "projections",
        "ignore_order",
        "limit",
        "page",
        "evaluation_id",
    }

    # Sync methods
    def list(
        self,
        query: Optional[Union[GetEventsQuery, Dict[str, Any]]] = None,
        *,
        data: Optional[Union[GetEventsQuery, Dict[str, Any]]] = None,
    ) -> GetEventsResponse:
        """Get events via the legacy list helper.

        Accepted keys:
            Project API key

        Args:
            query: Query parameters as GetEventsQuery model or dict.
                   Supported fields: dateRange, filters, projections,
                   ignore_order, limit, page, evaluation_id
            data: Backwards compatible alias for query.

        Returns:
            GetEventsResponse with matching events
        """
        # Support the legacy `data=` keyword while keeping the canonical `query=`
        # interface aligned with the current wrapper signature.
        payload = data if data is not None else query
        if payload is None:
            raise ValueError("EventsAPI.list requires query or data")

        # Convert to dict if Pydantic model
        if hasattr(payload, "model_dump"):
            payload_data = payload.model_dump(exclude_none=True)
        else:
            payload_data = payload

        # Filter data to only include supported parameters for the legacy list()
        # helper. The canonical spec exposes event reads through export().
        filtered_data = {
            k: v
            for k, v in payload_data.items()
            if k in self._GET_EVENTS_SUPPORTED_PARAMS
        }
        export_response = self.export(
            filters=filtered_data.get("filters"),
            date_range=filtered_data.get("dateRange"),
            projections=filtered_data.get("projections"),
            limit=filtered_data.get("limit", 1000),
            page=filtered_data.get("page", 1),
        )
        return GetEventsResponse(
            events=export_response.events,
            totalEvents=export_response.total_events,
        )

    @deprecated(
        "events.get_by_session_id() is deprecated; use events.export() with a "
        "session_id filter instead.",
        category=None,
    )
    def get_by_session_id(
        self,
        session_id: str,
        project: Optional[str] = None,
        *,
        limit: int = 1000,
    ) -> EventExportResponse:
        """Get events by session ID using the Data Plane export endpoint.

        Deprecated: this compatibility helper wraps export() with a session_id
        filter. New code should call events.export() directly.
        Events are returned sorted by start_time in chronological order.

        Accepted keys:
            Project API key

        Args:
            session_id: The session ID to fetch events for.
            project: Project name associated with the events. Deprecated in v1.0
                and will be removed in v2.0. The backend now infers project from
                session_id.
            limit: Maximum number of events to return (default 1000).

        Returns:
            EventExportResponse with events for the session, sorted by start_time.

        Example::

            response = client.events.export(
                filters=[
                    EventFilter(
                        field="session_id",
                        operator="is",
                        value="abc-123",
                        type="string",
                    )
                ]
            )
            for event in response.events:
                print(event["event_name"])
        """
        logger.debug(
            "get_by_session_id called: session_id=%s limit=%d",
            session_id,
            limit,
        )
        warnings.warn(
            "events.get_by_session_id() is deprecated and kept for backward "
            "compatibility only. Use events.export() with a session_id filter "
            "instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        if project is not None:
            warnings.warn(
                "The 'project' parameter is deprecated and will be removed in v2.0. "
                "The backend now infers project from session_id.",
                DeprecationWarning,
                stacklevel=2,
            )

        result = self.export(
            project=project,
            filters=[
                EventFilter(
                    field="session_id",
                    operator="is",
                    value=session_id,
                    type="string",
                )
            ],
            limit=limit,
            _sort_by_time=True,  # Enable time-based sorting
        )
        logger.debug(
            "get_by_session_id result: session_id=%s events=%d total=%d",
            session_id,
            len(result.events),
            result.total_events,
        )
        return result

    def create(self, request: PostEventRequest) -> PostEventResponse:
        """Create an event.

        Accepted keys:
            Ingestion API key; Project API key (deprecated)
        """
        return events_svc.createEventLegacy(self._ingestion_api_config, data=request)

    def update(self, data: UpdateEventRequest) -> None:
        """Update an event.

        Accepted keys:
            Ingestion API key; Project API key (deprecated)
        """
        return events_svc.updateEventLegacy(self._ingestion_api_config, data=data)

    def create_batch(self, data: PostEventBatchRequest) -> PostEventBatchResponse:
        """Create events in batch.

        Accepted keys:
            Ingestion API key; Project API key (deprecated)
        """
        return events_svc.createEventBatchLegacy(self._ingestion_api_config, data=data)

    def export(
        self,
        project: Optional[str] = None,
        filters: Optional[List[Union[EventFilter, Dict[str, Any]]]] = None,
        *,
        date_range: Optional[Dict[str, str]] = None,
        projections: Optional[List[str]] = None,
        limit: int = 1000,
        page: int = 1,
        _sort_by_time: bool = False,
    ) -> EventExportResponse:
        """Export events via POST /events/export (Data Plane).

        This is the primary method for retrieving events from HoneyHive.
        It uses the Data Plane endpoint which supports filtering by session_id,
        event_type, and other fields.

        Accepted keys:
            Project API key

        Args:
            project: Project name associated with the events. Deprecated in v1.0
                and will be removed in v2.0. The backend now infers project from
                filters (e.g., session_id).
            filters: List of EventFilter objects or dicts with filter criteria.
                Each filter should have: field, operator, value, type.
            date_range: Optional date range filter with '$gte' and '$lte' keys
                containing ISO timestamp strings.
            projections: Optional list of fields to include in the response.
            limit: Maximum number of results (default 1000, max 7500).
            page: Page number for pagination (default 1).
            _sort_by_time: Internal flag to sort events by start_time (default False).

        Returns:
            EventExportResponse with events list and total_events count.

        Example::

            from honeyhive.models import EventFilter

            # Export events for a session
            response = client.events.export(
                filters=[
                    EventFilter(
                        field="session_id",
                        operator="is",
                        value="abc-123",
                        type="string"
                    )
                ],
                limit=100
            )

            for event in response.events:
                print(event["event_name"])

            # Export with date range
            response = client.events.export(
                filters=[],
                date_range={
                    "$gte": "2024-01-01T00:00:00Z",
                    "$lte": "2024-01-31T23:59:59Z"
                }
            )
        """
        if project is not None:
            warnings.warn(
                "The 'project' parameter is deprecated and will be removed in v2.0. "
                "The backend now infers project from filters (e.g., session_id).",
                DeprecationWarning,
                stacklevel=2,
            )

        # Build filters array
        filters_data = []
        if filters:
            for f in filters:
                if isinstance(f, EventFilter):
                    filters_data.append(f.to_dict())
                elif isinstance(f, dict):
                    filters_data.append(f)

        # Build request body
        request_body: Dict[str, Any] = {
            "filters": filters_data,
            "limit": limit,
            "page": page,
        }

        # Only include project if provided (for backwards compatibility)
        if project is not None:
            request_body["project"] = project

        if date_range:
            request_body["dateRange"] = date_range
        if projections:
            request_body["projections"] = projections

        # Make direct request to /events/export (bypasses generated model issues)
        api_config = self._api_config
        base_path = api_config.base_path
        headers = api_config.get_default_headers()

        # Log outgoing request metadata
        logger.debug(
            "export request: POST %s/v1/events/export limit=%d page=%d",
            base_path,
            limit,
            page,
        )

        # Retry 408/429/500/502/503/504 and transport timeouts. Budget comes
        # from HoneyHive(retry_config=...) or HH_MAX_RETRIES. The other API
        # methods do not retry.
        with httpx.Client(
            base_url=base_path,
            verify=api_config.verify,
            timeout=EXPORT_TIMEOUT,
        ) as client:
            response = self._retry_config.execute(
                lambda: client.request(
                    "POST",
                    "/v1/events/export",
                    headers=headers,
                    json=request_body,
                ),
                operation="export()",
            )

        data = response.json()
        events = data.get("events", [])
        total_events = data.get("totalEvents", data.get("count", 0))

        logger.debug(
            "export result: %d events, total_events=%d",
            len(events),
            total_events,
        )
        if not events:
            logger.debug(
                "export returned empty events list; limit=%d page=%d",
                limit,
                page,
            )

        # Sort events by start_time if requested (fixes jumbled order issue)
        if _sort_by_time and events:
            events = self._sort_events_by_time(events)

        return EventExportResponse(
            events=events,
            total_events=total_events,
        )

    def _sort_events_by_time(
        self, events: List[Dict[str, Any]]
    ) -> List[Dict[str, Any]]:
        """Sort events by start_time in chronological order.

        Args:
            events: List of event dictionaries.

        Returns:
            Sorted list of events by start_time.
        """

        def get_start_time(event: Dict[str, Any]) -> float:
            """Extract start_time from event, handling various formats."""
            # Try different possible field names for start time
            start_time = (
                event.get("start_time")
                or event.get("startTime")
                or event.get("created_at")
            )
            if start_time is None:
                return 0.0
            # Handle both numeric timestamps and ISO string formats
            if isinstance(start_time, (int, float)):
                return float(start_time)
            if isinstance(start_time, str):
                try:
                    from datetime import datetime

                    # Try ISO format parsing
                    if "T" in start_time:
                        dt = datetime.fromisoformat(start_time.replace("Z", "+00:00"))
                        return dt.timestamp()
                except (ValueError, TypeError):
                    pass
            return 0.0

        return sorted(events, key=get_start_time)

    # Async methods
    async def list_async(
        self, query: Union[GetEventsQuery, Dict[str, Any]]
    ) -> GetEventsResponse:
        """Get events asynchronously via the legacy list helper.

        Accepted keys:
            Project API key

        Args:
            query: Query parameters as GetEventsQuery model or dict.

        Returns:
            GetEventsResponse with matching events
        """
        # Convert to dict if Pydantic model
        if hasattr(query, "model_dump"):
            data = query.model_dump(exclude_none=True)
        else:
            data = query

        # Filter data to only include supported parameters for the legacy list()
        # helper. The canonical spec exposes event reads through export().
        filtered_data = {
            k: v for k, v in data.items() if k in self._GET_EVENTS_SUPPORTED_PARAMS
        }
        export_response = await self.export_async(
            filters=filtered_data.get("filters"),
            date_range=filtered_data.get("dateRange"),
            projections=filtered_data.get("projections"),
            limit=filtered_data.get("limit", 1000),
            page=filtered_data.get("page", 1),
        )
        return GetEventsResponse(
            events=export_response.events,
            totalEvents=export_response.total_events,
        )

    @deprecated(
        "events.get_by_session_id_async() is deprecated; use "
        "events.export_async() with a session_id filter instead.",
        category=None,
    )
    async def get_by_session_id_async(
        self,
        session_id: str,
        project: Optional[str] = None,
        *,
        limit: int = 1000,
    ) -> EventExportResponse:
        """Get events by session ID asynchronously using the Data Plane export endpoint.

        Deprecated: this compatibility helper wraps export_async() with a
        session_id filter. New code should call events.export_async() directly.
        Events are returned sorted by start_time in chronological order.

        Accepted keys:
            Project API key

        Args:
            session_id: The session ID to fetch events for.
            project: Project name associated with the events. Deprecated in v1.0
                and will be removed in v2.0. The backend now infers project from
                session_id.
            limit: Maximum number of events to return (default 1000).

        Returns:
            EventExportResponse with events for the session, sorted by start_time.
        """
        logger.debug(
            "get_by_session_id_async called: session_id=%s limit=%d",
            session_id,
            limit,
        )
        warnings.warn(
            "events.get_by_session_id_async() is deprecated and kept for backward "
            "compatibility only. Use events.export_async() with a session_id "
            "filter instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        if project is not None:
            warnings.warn(
                "The 'project' parameter is deprecated and will be removed in v2.0. "
                "The backend now infers project from session_id.",
                DeprecationWarning,
                stacklevel=2,
            )

        result = await self.export_async(
            project=project,
            filters=[
                EventFilter(
                    field="session_id",
                    operator="is",
                    value=session_id,
                    type="string",
                )
            ],
            limit=limit,
            _sort_by_time=True,  # Enable time-based sorting
        )
        logger.debug(
            "get_by_session_id_async result: session_id=%s events=%d total=%d",
            session_id,
            len(result.events),
            result.total_events,
        )
        return result

    async def create_async(self, request: PostEventRequest) -> PostEventResponse:
        """Create an event asynchronously.

        Accepted keys:
            Ingestion API key; Project API key (deprecated)
        """
        return await events_svc_async.createEventLegacy(
            self._ingestion_api_config, data=request
        )

    async def update_async(self, data: UpdateEventRequest) -> None:
        """Update an event asynchronously.

        Accepted keys:
            Ingestion API key; Project API key (deprecated)
        """
        return await events_svc_async.updateEventLegacy(
            self._ingestion_api_config, data=data
        )

    async def create_batch_async(
        self, data: PostEventBatchRequest
    ) -> PostEventBatchResponse:
        """Create events in batch asynchronously.

        Accepted keys:
            Ingestion API key; Project API key (deprecated)
        """
        return await events_svc_async.createEventBatchLegacy(
            self._ingestion_api_config, data=data
        )

    async def export_async(
        self,
        project: Optional[str] = None,
        filters: Optional[List[Union[EventFilter, Dict[str, Any]]]] = None,
        *,
        date_range: Optional[Dict[str, str]] = None,
        projections: Optional[List[str]] = None,
        limit: int = 1000,
        page: int = 1,
        _sort_by_time: bool = False,
    ) -> EventExportResponse:
        """Export events via POST /events/export asynchronously (Data Plane).

        Async version of export(). See export() for full documentation.

        Accepted keys:
            Project API key

        Args:
            project: Project name associated with the events. Deprecated in v1.0
                and will be removed in v2.0. The backend now infers project from
                filters (e.g., session_id).
            filters: List of EventFilter objects or dicts with filter criteria.
            date_range: Optional date range filter.
            projections: Optional list of fields to include in the response.
            limit: Maximum number of results (default 1000, max 7500).
            page: Page number for pagination (default 1).
            _sort_by_time: Internal flag to sort events by start_time (default False).

        Returns:
            EventExportResponse with events list and total_events count.
        """
        if project is not None:
            warnings.warn(
                "The 'project' parameter is deprecated and will be removed in v2.0. "
                "The backend now infers project from filters (e.g., session_id).",
                DeprecationWarning,
                stacklevel=2,
            )

        # Build filters array
        filters_data = []
        if filters:
            for f in filters:
                if isinstance(f, EventFilter):
                    filters_data.append(f.to_dict())
                elif isinstance(f, dict):
                    filters_data.append(f)

        # Build request body
        request_body: Dict[str, Any] = {
            "filters": filters_data,
            "limit": limit,
            "page": page,
        }

        # Only include project if provided (for backwards compatibility)
        if project is not None:
            request_body["project"] = project

        if date_range:
            request_body["dateRange"] = date_range
        if projections:
            request_body["projections"] = projections

        # Make direct async request to /events/export
        api_config = self._api_config
        base_path = api_config.base_path
        headers = api_config.get_default_headers()

        # Log outgoing request metadata
        logger.debug(
            "export_async request: POST %s/v1/events/export limit=%d page=%d",
            base_path,
            limit,
            page,
        )

        # Retry 408/429/500/502/503/504 and transport timeouts. Budget comes
        # from HoneyHive(retry_config=...) or HH_MAX_RETRIES. The other API
        # methods do not retry.
        async with httpx.AsyncClient(
            base_url=base_path,
            verify=api_config.verify,
            timeout=EXPORT_TIMEOUT,
        ) as client:
            response = await self._retry_config.execute_async(
                lambda: client.request(
                    "POST",
                    "/v1/events/export",
                    headers=headers,
                    json=request_body,
                ),
                operation="export_async()",
            )

        data = response.json()
        events = data.get("events", [])
        total_events = data.get("totalEvents", data.get("count", 0))

        logger.debug(
            "export_async result: %d events, total_events=%d",
            len(events),
            total_events,
        )
        if not events:
            logger.debug(
                "export_async returned empty events list; limit=%d page=%d",
                limit,
                page,
            )

        # Sort events by start_time if requested (fixes jumbled order issue)
        if _sort_by_time and events:
            events = self._sort_events_by_time(events)

        return EventExportResponse(
            events=events,
            total_events=total_events,
        )

    # Backwards compatible aliases
    def create_event(self, request: PostEventRequest) -> PostEventResponse:
        """Create an event (backwards compatible alias for create()).

        Accepted keys:
            Ingestion API key; Project API key (deprecated)
        """
        return self.create(request)

    def update_event(self, data: Union[UpdateEventRequest, Dict[str, Any]]) -> None:
        """Update an event (backwards compatible alias for update()).

        Accepts the request model or the equivalent dict. A dict is validated
        into the model, because ``update()`` sends a model.

        Accepted keys:
            Ingestion API key; Project API key (deprecated)
        """
        if not isinstance(data, UpdateEventRequest):
            data = UpdateEventRequest.model_validate(data)
        return self.update(data)

    def list_events(
        self,
        project: Optional[str] = None,
        filters: Optional[List[Union[EventFilter, Dict[str, Any]]]] = None,
        *,
        date_range: Optional[Dict[str, str]] = None,
        projections: Optional[List[str]] = None,
        limit: int = 1000,
        page: int = 1,
    ) -> EventExportResponse:
        """List events via export endpoint (backwards compatible alias).

        This is a backwards compatible alias for export(). Uses the Data Plane
        POST /events/export endpoint.

        Accepted keys:
            Project API key

        Args:
            project: Project name. Deprecated in v1.0 and will be removed in v2.0.
            filters: List of EventFilter objects or dicts.
            date_range: Optional date range filter.
            projections: Optional list of fields to include.
            limit: Maximum number of results (default 1000).
            page: Page number (default 1).

        Returns:
            EventExportResponse with events and total count.
        """
        return self.export(
            project=project,
            filters=filters,
            date_range=date_range,
            projections=projections,
            limit=limit,
            page=page,
        )

    def get_events(
        self,
        project: Optional[str] = None,
        filters: Optional[List[Union[EventFilter, Dict[str, Any]]]] = None,
        *,
        date_range: Optional[Dict[str, str]] = None,
        projections: Optional[List[str]] = None,
        limit: int = 1000,
        page: int = 1,
    ) -> EventExportResponse:
        """Get events via export endpoint (backwards compatible alias).

        This is a backwards compatible alias for export(). Uses the Data Plane
        POST /events/export endpoint.

        Accepted keys:
            Project API key

        Args:
            project: Project name. Deprecated in v1.0 and will be removed in v2.0.
            filters: List of EventFilter objects or dicts.
            date_range: Optional date range filter.
            projections: Optional list of fields to include.
            limit: Maximum number of results (default 1000).
            page: Page number (default 1).

        Returns:
            EventExportResponse with events and total count.
        """
        return self.export(
            project=project,
            filters=filters,
            date_range=date_range,
            projections=projections,
            limit=limit,
            page=page,
        )

    async def list_events_async(
        self,
        project: Optional[str] = None,
        filters: Optional[List[Union[EventFilter, Dict[str, Any]]]] = None,
        *,
        date_range: Optional[Dict[str, str]] = None,
        projections: Optional[List[str]] = None,
        limit: int = 1000,
        page: int = 1,
    ) -> EventExportResponse:
        """List events asynchronously (backwards compatible alias for export_async).

        Accepted keys:
            Project API key

        Args:
            project: Project name. Deprecated in v1.0 and will be removed in v2.0.
        """
        return await self.export_async(
            project=project,
            filters=filters,
            date_range=date_range,
            projections=projections,
            limit=limit,
            page=page,
        )

    async def get_events_async(
        self,
        project: Optional[str] = None,
        filters: Optional[List[Union[EventFilter, Dict[str, Any]]]] = None,
        *,
        date_range: Optional[Dict[str, str]] = None,
        projections: Optional[List[str]] = None,
        limit: int = 1000,
        page: int = 1,
    ) -> EventExportResponse:
        """Get events asynchronously (backwards compatible alias for export_async).

        Accepted keys:
            Project API key

        Args:
            project: Project name. Deprecated in v1.0 and will be removed in v2.0.
        """
        return await self.export_async(
            project=project,
            filters=filters,
            date_range=date_range,
            projections=projections,
            limit=limit,
            page=page,
        )

list

list(
    query: Optional[
        Union[GetEventsQuery, Dict[str, Any]]
    ] = None,
    *,
    data: Optional[
        Union[GetEventsQuery, Dict[str, Any]]
    ] = None
) -> GetEventsResponse

Get events via the legacy list helper.

Accepted keys

Project API key

Parameters:

Name	Type	Description	Default
query	Optional[Union[GetEventsQuery, Dict[str, Any]]]	Query parameters as GetEventsQuery model or dict. Supported fields: dateRange, filters, projections, ignore_order, limit, page, evaluation_id	None
data	Optional[Union[GetEventsQuery, Dict[str, Any]]]	Backwards compatible alias for query.	None

Returns:

Type	Description
GetEventsResponse	GetEventsResponse with matching events

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

def list(
    self,
    query: Optional[Union[GetEventsQuery, Dict[str, Any]]] = None,
    *,
    data: Optional[Union[GetEventsQuery, Dict[str, Any]]] = None,
) -> GetEventsResponse:
    """Get events via the legacy list helper.

    Accepted keys:
        Project API key

    Args:
        query: Query parameters as GetEventsQuery model or dict.
               Supported fields: dateRange, filters, projections,
               ignore_order, limit, page, evaluation_id
        data: Backwards compatible alias for query.

    Returns:
        GetEventsResponse with matching events
    """
    # Support the legacy `data=` keyword while keeping the canonical `query=`
    # interface aligned with the current wrapper signature.
    payload = data if data is not None else query
    if payload is None:
        raise ValueError("EventsAPI.list requires query or data")

    # Convert to dict if Pydantic model
    if hasattr(payload, "model_dump"):
        payload_data = payload.model_dump(exclude_none=True)
    else:
        payload_data = payload

    # Filter data to only include supported parameters for the legacy list()
    # helper. The canonical spec exposes event reads through export().
    filtered_data = {
        k: v
        for k, v in payload_data.items()
        if k in self._GET_EVENTS_SUPPORTED_PARAMS
    }
    export_response = self.export(
        filters=filtered_data.get("filters"),
        date_range=filtered_data.get("dateRange"),
        projections=filtered_data.get("projections"),
        limit=filtered_data.get("limit", 1000),
        page=filtered_data.get("page", 1),
    )
    return GetEventsResponse(
        events=export_response.events,
        totalEvents=export_response.total_events,
    )

get_by_session_id

get_by_session_id(
    session_id: str,
    project: Optional[str] = None,
    *,
    limit: int = 1000
) -> EventExportResponse

Get events by session ID using the Data Plane export endpoint.

Deprecated: this compatibility helper wraps export() with a session_id filter. New code should call events.export() directly. Events are returned sorted by start_time in chronological order.

Accepted keys

Project API key

Parameters:

Name	Type	Description	Default
session_id	str	The session ID to fetch events for.	required
project	Optional[str]	Project name associated with the events. Deprecated in v1.0 and will be removed in v2.0. The backend now infers project from session_id.	None
limit	int	Maximum number of events to return (default 1000).	1000

Returns:

Type	Description
EventExportResponse	EventExportResponse with events for the session, sorted by start_time.

Example::

response = client.events.export(
    filters=[
        EventFilter(
            field="session_id",
            operator="is",
            value="abc-123",
            type="string",
        )
    ]
)
for event in response.events:
    print(event["event_name"])

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

@deprecated(
    "events.get_by_session_id() is deprecated; use events.export() with a "
    "session_id filter instead.",
    category=None,
)
def get_by_session_id(
    self,
    session_id: str,
    project: Optional[str] = None,
    *,
    limit: int = 1000,
) -> EventExportResponse:
    """Get events by session ID using the Data Plane export endpoint.

    Deprecated: this compatibility helper wraps export() with a session_id
    filter. New code should call events.export() directly.
    Events are returned sorted by start_time in chronological order.

    Accepted keys:
        Project API key

    Args:
        session_id: The session ID to fetch events for.
        project: Project name associated with the events. Deprecated in v1.0
            and will be removed in v2.0. The backend now infers project from
            session_id.
        limit: Maximum number of events to return (default 1000).

    Returns:
        EventExportResponse with events for the session, sorted by start_time.

    Example::

        response = client.events.export(
            filters=[
                EventFilter(
                    field="session_id",
                    operator="is",
                    value="abc-123",
                    type="string",
                )
            ]
        )
        for event in response.events:
            print(event["event_name"])
    """
    logger.debug(
        "get_by_session_id called: session_id=%s limit=%d",
        session_id,
        limit,
    )
    warnings.warn(
        "events.get_by_session_id() is deprecated and kept for backward "
        "compatibility only. Use events.export() with a session_id filter "
        "instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    if project is not None:
        warnings.warn(
            "The 'project' parameter is deprecated and will be removed in v2.0. "
            "The backend now infers project from session_id.",
            DeprecationWarning,
            stacklevel=2,
        )

    result = self.export(
        project=project,
        filters=[
            EventFilter(
                field="session_id",
                operator="is",
                value=session_id,
                type="string",
            )
        ],
        limit=limit,
        _sort_by_time=True,  # Enable time-based sorting
    )
    logger.debug(
        "get_by_session_id result: session_id=%s events=%d total=%d",
        session_id,
        len(result.events),
        result.total_events,
    )
    return result

create

create(request: PostEventRequest) -> PostEventResponse

Create an event.

Accepted keys

Ingestion API key; Project API key (deprecated)

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

def create(self, request: PostEventRequest) -> PostEventResponse:
    """Create an event.

    Accepted keys:
        Ingestion API key; Project API key (deprecated)
    """
    return events_svc.createEventLegacy(self._ingestion_api_config, data=request)

update

update(data: UpdateEventRequest) -> None

Update an event.

Accepted keys

Ingestion API key; Project API key (deprecated)

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

def update(self, data: UpdateEventRequest) -> None:
    """Update an event.

    Accepted keys:
        Ingestion API key; Project API key (deprecated)
    """
    return events_svc.updateEventLegacy(self._ingestion_api_config, data=data)

create_batch

create_batch(
    data: PostEventBatchRequest,
) -> PostEventBatchResponse

Create events in batch.

Accepted keys

Ingestion API key; Project API key (deprecated)

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

def create_batch(self, data: PostEventBatchRequest) -> PostEventBatchResponse:
    """Create events in batch.

    Accepted keys:
        Ingestion API key; Project API key (deprecated)
    """
    return events_svc.createEventBatchLegacy(self._ingestion_api_config, data=data)

export

export(
    project: Optional[str] = None,
    filters: Optional[
        List[Union[EventFilter, Dict[str, Any]]]
    ] = None,
    *,
    date_range: Optional[Dict[str, str]] = None,
    projections: Optional[List[str]] = None,
    limit: int = 1000,
    page: int = 1,
    _sort_by_time: bool = False
) -> EventExportResponse

Export events via POST /events/export (Data Plane).

This is the primary method for retrieving events from HoneyHive. It uses the Data Plane endpoint which supports filtering by session_id, event_type, and other fields.

Accepted keys

Project API key

Parameters:

Name	Type	Description	Default
project	Optional[str]	Project name associated with the events. Deprecated in v1.0 and will be removed in v2.0. The backend now infers project from filters (e.g., session_id).	None
filters	Optional[List[Union[EventFilter, Dict[str, Any]]]]	List of EventFilter objects or dicts with filter criteria. Each filter should have: field, operator, value, type.	None
date_range	Optional[Dict[str, str]]	Optional date range filter with '$gte' and '$lte' keys containing ISO timestamp strings.	None
projections	Optional[List[str]]	Optional list of fields to include in the response.	None
limit	int	Maximum number of results (default 1000, max 7500).	1000
page	int	Page number for pagination (default 1).	1
_sort_by_time	bool	Internal flag to sort events by start_time (default False).	False

Returns:

Type	Description
EventExportResponse	EventExportResponse with events list and total_events count.

Example::

from honeyhive.models import EventFilter

# Export events for a session
response = client.events.export(
    filters=[
        EventFilter(
            field="session_id",
            operator="is",
            value="abc-123",
            type="string"
        )
    ],
    limit=100
)

for event in response.events:
    print(event["event_name"])

# Export with date range
response = client.events.export(
    filters=[],
    date_range={
        "$gte": "2024-01-01T00:00:00Z",
        "$lte": "2024-01-31T23:59:59Z"
    }
)

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

def export(
    self,
    project: Optional[str] = None,
    filters: Optional[List[Union[EventFilter, Dict[str, Any]]]] = None,
    *,
    date_range: Optional[Dict[str, str]] = None,
    projections: Optional[List[str]] = None,
    limit: int = 1000,
    page: int = 1,
    _sort_by_time: bool = False,
) -> EventExportResponse:
    """Export events via POST /events/export (Data Plane).

    This is the primary method for retrieving events from HoneyHive.
    It uses the Data Plane endpoint which supports filtering by session_id,
    event_type, and other fields.

    Accepted keys:
        Project API key

    Args:
        project: Project name associated with the events. Deprecated in v1.0
            and will be removed in v2.0. The backend now infers project from
            filters (e.g., session_id).
        filters: List of EventFilter objects or dicts with filter criteria.
            Each filter should have: field, operator, value, type.
        date_range: Optional date range filter with '$gte' and '$lte' keys
            containing ISO timestamp strings.
        projections: Optional list of fields to include in the response.
        limit: Maximum number of results (default 1000, max 7500).
        page: Page number for pagination (default 1).
        _sort_by_time: Internal flag to sort events by start_time (default False).

    Returns:
        EventExportResponse with events list and total_events count.

    Example::

        from honeyhive.models import EventFilter

        # Export events for a session
        response = client.events.export(
            filters=[
                EventFilter(
                    field="session_id",
                    operator="is",
                    value="abc-123",
                    type="string"
                )
            ],
            limit=100
        )

        for event in response.events:
            print(event["event_name"])

        # Export with date range
        response = client.events.export(
            filters=[],
            date_range={
                "$gte": "2024-01-01T00:00:00Z",
                "$lte": "2024-01-31T23:59:59Z"
            }
        )
    """
    if project is not None:
        warnings.warn(
            "The 'project' parameter is deprecated and will be removed in v2.0. "
            "The backend now infers project from filters (e.g., session_id).",
            DeprecationWarning,
            stacklevel=2,
        )

    # Build filters array
    filters_data = []
    if filters:
        for f in filters:
            if isinstance(f, EventFilter):
                filters_data.append(f.to_dict())
            elif isinstance(f, dict):
                filters_data.append(f)

    # Build request body
    request_body: Dict[str, Any] = {
        "filters": filters_data,
        "limit": limit,
        "page": page,
    }

    # Only include project if provided (for backwards compatibility)
    if project is not None:
        request_body["project"] = project

    if date_range:
        request_body["dateRange"] = date_range
    if projections:
        request_body["projections"] = projections

    # Make direct request to /events/export (bypasses generated model issues)
    api_config = self._api_config
    base_path = api_config.base_path
    headers = api_config.get_default_headers()

    # Log outgoing request metadata
    logger.debug(
        "export request: POST %s/v1/events/export limit=%d page=%d",
        base_path,
        limit,
        page,
    )

    # Retry 408/429/500/502/503/504 and transport timeouts. Budget comes
    # from HoneyHive(retry_config=...) or HH_MAX_RETRIES. The other API
    # methods do not retry.
    with httpx.Client(
        base_url=base_path,
        verify=api_config.verify,
        timeout=EXPORT_TIMEOUT,
    ) as client:
        response = self._retry_config.execute(
            lambda: client.request(
                "POST",
                "/v1/events/export",
                headers=headers,
                json=request_body,
            ),
            operation="export()",
        )

    data = response.json()
    events = data.get("events", [])
    total_events = data.get("totalEvents", data.get("count", 0))

    logger.debug(
        "export result: %d events, total_events=%d",
        len(events),
        total_events,
    )
    if not events:
        logger.debug(
            "export returned empty events list; limit=%d page=%d",
            limit,
            page,
        )

    # Sort events by start_time if requested (fixes jumbled order issue)
    if _sort_by_time and events:
        events = self._sort_events_by_time(events)

    return EventExportResponse(
        events=events,
        total_events=total_events,
    )

list_async async

list_async(
    query: Union[GetEventsQuery, Dict[str, Any]],
) -> GetEventsResponse

Get events asynchronously via the legacy list helper.

Accepted keys

Project API key

Parameters:

Name	Type	Description	Default
query	Union[GetEventsQuery, Dict[str, Any]]	Query parameters as GetEventsQuery model or dict.	required

Returns:

Type	Description
GetEventsResponse	GetEventsResponse with matching events

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

async def list_async(
    self, query: Union[GetEventsQuery, Dict[str, Any]]
) -> GetEventsResponse:
    """Get events asynchronously via the legacy list helper.

    Accepted keys:
        Project API key

    Args:
        query: Query parameters as GetEventsQuery model or dict.

    Returns:
        GetEventsResponse with matching events
    """
    # Convert to dict if Pydantic model
    if hasattr(query, "model_dump"):
        data = query.model_dump(exclude_none=True)
    else:
        data = query

    # Filter data to only include supported parameters for the legacy list()
    # helper. The canonical spec exposes event reads through export().
    filtered_data = {
        k: v for k, v in data.items() if k in self._GET_EVENTS_SUPPORTED_PARAMS
    }
    export_response = await self.export_async(
        filters=filtered_data.get("filters"),
        date_range=filtered_data.get("dateRange"),
        projections=filtered_data.get("projections"),
        limit=filtered_data.get("limit", 1000),
        page=filtered_data.get("page", 1),
    )
    return GetEventsResponse(
        events=export_response.events,
        totalEvents=export_response.total_events,
    )

get_by_session_id_async async

get_by_session_id_async(
    session_id: str,
    project: Optional[str] = None,
    *,
    limit: int = 1000
) -> EventExportResponse

Get events by session ID asynchronously using the Data Plane export endpoint.

Deprecated: this compatibility helper wraps export_async() with a session_id filter. New code should call events.export_async() directly. Events are returned sorted by start_time in chronological order.

Accepted keys

Project API key

Parameters:

Name	Type	Description	Default
session_id	str	The session ID to fetch events for.	required
project	Optional[str]	Project name associated with the events. Deprecated in v1.0 and will be removed in v2.0. The backend now infers project from session_id.	None
limit	int	Maximum number of events to return (default 1000).	1000

Returns:

Type	Description
EventExportResponse	EventExportResponse with events for the session, sorted by start_time.

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

@deprecated(
    "events.get_by_session_id_async() is deprecated; use "
    "events.export_async() with a session_id filter instead.",
    category=None,
)
async def get_by_session_id_async(
    self,
    session_id: str,
    project: Optional[str] = None,
    *,
    limit: int = 1000,
) -> EventExportResponse:
    """Get events by session ID asynchronously using the Data Plane export endpoint.

    Deprecated: this compatibility helper wraps export_async() with a
    session_id filter. New code should call events.export_async() directly.
    Events are returned sorted by start_time in chronological order.

    Accepted keys:
        Project API key

    Args:
        session_id: The session ID to fetch events for.
        project: Project name associated with the events. Deprecated in v1.0
            and will be removed in v2.0. The backend now infers project from
            session_id.
        limit: Maximum number of events to return (default 1000).

    Returns:
        EventExportResponse with events for the session, sorted by start_time.
    """
    logger.debug(
        "get_by_session_id_async called: session_id=%s limit=%d",
        session_id,
        limit,
    )
    warnings.warn(
        "events.get_by_session_id_async() is deprecated and kept for backward "
        "compatibility only. Use events.export_async() with a session_id "
        "filter instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    if project is not None:
        warnings.warn(
            "The 'project' parameter is deprecated and will be removed in v2.0. "
            "The backend now infers project from session_id.",
            DeprecationWarning,
            stacklevel=2,
        )

    result = await self.export_async(
        project=project,
        filters=[
            EventFilter(
                field="session_id",
                operator="is",
                value=session_id,
                type="string",
            )
        ],
        limit=limit,
        _sort_by_time=True,  # Enable time-based sorting
    )
    logger.debug(
        "get_by_session_id_async result: session_id=%s events=%d total=%d",
        session_id,
        len(result.events),
        result.total_events,
    )
    return result

create_async async

create_async(
    request: PostEventRequest,
) -> PostEventResponse

Create an event asynchronously.

Accepted keys

Ingestion API key; Project API key (deprecated)

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

async def create_async(self, request: PostEventRequest) -> PostEventResponse:
    """Create an event asynchronously.

    Accepted keys:
        Ingestion API key; Project API key (deprecated)
    """
    return await events_svc_async.createEventLegacy(
        self._ingestion_api_config, data=request
    )

update_async async

update_async(data: UpdateEventRequest) -> None

Update an event asynchronously.

Accepted keys

Ingestion API key; Project API key (deprecated)

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

async def update_async(self, data: UpdateEventRequest) -> None:
    """Update an event asynchronously.

    Accepted keys:
        Ingestion API key; Project API key (deprecated)
    """
    return await events_svc_async.updateEventLegacy(
        self._ingestion_api_config, data=data
    )

create_batch_async async

create_batch_async(
    data: PostEventBatchRequest,
) -> PostEventBatchResponse

Create events in batch asynchronously.

Accepted keys

Ingestion API key; Project API key (deprecated)

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

async def create_batch_async(
    self, data: PostEventBatchRequest
) -> PostEventBatchResponse:
    """Create events in batch asynchronously.

    Accepted keys:
        Ingestion API key; Project API key (deprecated)
    """
    return await events_svc_async.createEventBatchLegacy(
        self._ingestion_api_config, data=data
    )

export_async async

export_async(
    project: Optional[str] = None,
    filters: Optional[
        List[Union[EventFilter, Dict[str, Any]]]
    ] = None,
    *,
    date_range: Optional[Dict[str, str]] = None,
    projections: Optional[List[str]] = None,
    limit: int = 1000,
    page: int = 1,
    _sort_by_time: bool = False
) -> EventExportResponse

Export events via POST /events/export asynchronously (Data Plane).

Async version of export(). See export() for full documentation.

Accepted keys

Project API key

Parameters:

Name	Type	Description	Default
project	Optional[str]	Project name associated with the events. Deprecated in v1.0 and will be removed in v2.0. The backend now infers project from filters (e.g., session_id).	None
filters	Optional[List[Union[EventFilter, Dict[str, Any]]]]	List of EventFilter objects or dicts with filter criteria.	None
date_range	Optional[Dict[str, str]]	Optional date range filter.	None
projections	Optional[List[str]]	Optional list of fields to include in the response.	None
limit	int	Maximum number of results (default 1000, max 7500).	1000
page	int	Page number for pagination (default 1).	1
_sort_by_time	bool	Internal flag to sort events by start_time (default False).	False

Returns:

Type	Description
EventExportResponse	EventExportResponse with events list and total_events count.

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

async def export_async(
    self,
    project: Optional[str] = None,
    filters: Optional[List[Union[EventFilter, Dict[str, Any]]]] = None,
    *,
    date_range: Optional[Dict[str, str]] = None,
    projections: Optional[List[str]] = None,
    limit: int = 1000,
    page: int = 1,
    _sort_by_time: bool = False,
) -> EventExportResponse:
    """Export events via POST /events/export asynchronously (Data Plane).

    Async version of export(). See export() for full documentation.

    Accepted keys:
        Project API key

    Args:
        project: Project name associated with the events. Deprecated in v1.0
            and will be removed in v2.0. The backend now infers project from
            filters (e.g., session_id).
        filters: List of EventFilter objects or dicts with filter criteria.
        date_range: Optional date range filter.
        projections: Optional list of fields to include in the response.
        limit: Maximum number of results (default 1000, max 7500).
        page: Page number for pagination (default 1).
        _sort_by_time: Internal flag to sort events by start_time (default False).

    Returns:
        EventExportResponse with events list and total_events count.
    """
    if project is not None:
        warnings.warn(
            "The 'project' parameter is deprecated and will be removed in v2.0. "
            "The backend now infers project from filters (e.g., session_id).",
            DeprecationWarning,
            stacklevel=2,
        )

    # Build filters array
    filters_data = []
    if filters:
        for f in filters:
            if isinstance(f, EventFilter):
                filters_data.append(f.to_dict())
            elif isinstance(f, dict):
                filters_data.append(f)

    # Build request body
    request_body: Dict[str, Any] = {
        "filters": filters_data,
        "limit": limit,
        "page": page,
    }

    # Only include project if provided (for backwards compatibility)
    if project is not None:
        request_body["project"] = project

    if date_range:
        request_body["dateRange"] = date_range
    if projections:
        request_body["projections"] = projections

    # Make direct async request to /events/export
    api_config = self._api_config
    base_path = api_config.base_path
    headers = api_config.get_default_headers()

    # Log outgoing request metadata
    logger.debug(
        "export_async request: POST %s/v1/events/export limit=%d page=%d",
        base_path,
        limit,
        page,
    )

    # Retry 408/429/500/502/503/504 and transport timeouts. Budget comes
    # from HoneyHive(retry_config=...) or HH_MAX_RETRIES. The other API
    # methods do not retry.
    async with httpx.AsyncClient(
        base_url=base_path,
        verify=api_config.verify,
        timeout=EXPORT_TIMEOUT,
    ) as client:
        response = await self._retry_config.execute_async(
            lambda: client.request(
                "POST",
                "/v1/events/export",
                headers=headers,
                json=request_body,
            ),
            operation="export_async()",
        )

    data = response.json()
    events = data.get("events", [])
    total_events = data.get("totalEvents", data.get("count", 0))

    logger.debug(
        "export_async result: %d events, total_events=%d",
        len(events),
        total_events,
    )
    if not events:
        logger.debug(
            "export_async returned empty events list; limit=%d page=%d",
            limit,
            page,
        )

    # Sort events by start_time if requested (fixes jumbled order issue)
    if _sort_by_time and events:
        events = self._sort_events_by_time(events)

    return EventExportResponse(
        events=events,
        total_events=total_events,
    )

create_event

create_event(
    request: PostEventRequest,
) -> PostEventResponse

Create an event (backwards compatible alias for create()).

Accepted keys

Ingestion API key; Project API key (deprecated)

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

def create_event(self, request: PostEventRequest) -> PostEventResponse:
    """Create an event (backwards compatible alias for create()).

    Accepted keys:
        Ingestion API key; Project API key (deprecated)
    """
    return self.create(request)

update_event

update_event(
    data: Union[UpdateEventRequest, Dict[str, Any]],
) -> None

Update an event (backwards compatible alias for update()).

Accepts the request model or the equivalent dict. A dict is validated into the model, because update() sends a model.

Accepted keys

Ingestion API key; Project API key (deprecated)

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

def update_event(self, data: Union[UpdateEventRequest, Dict[str, Any]]) -> None:
    """Update an event (backwards compatible alias for update()).

    Accepts the request model or the equivalent dict. A dict is validated
    into the model, because ``update()`` sends a model.

    Accepted keys:
        Ingestion API key; Project API key (deprecated)
    """
    if not isinstance(data, UpdateEventRequest):
        data = UpdateEventRequest.model_validate(data)
    return self.update(data)

list_events

list_events(
    project: Optional[str] = None,
    filters: Optional[
        List[Union[EventFilter, Dict[str, Any]]]
    ] = None,
    *,
    date_range: Optional[Dict[str, str]] = None,
    projections: Optional[List[str]] = None,
    limit: int = 1000,
    page: int = 1
) -> EventExportResponse

List events via export endpoint (backwards compatible alias).

This is a backwards compatible alias for export(). Uses the Data Plane POST /events/export endpoint.

Accepted keys

Project API key

Parameters:

Name	Type	Description	Default
project	Optional[str]	Project name. Deprecated in v1.0 and will be removed in v2.0.	None
filters	Optional[List[Union[EventFilter, Dict[str, Any]]]]	List of EventFilter objects or dicts.	None
date_range	Optional[Dict[str, str]]	Optional date range filter.	None
projections	Optional[List[str]]	Optional list of fields to include.	None
limit	int	Maximum number of results (default 1000).	1000
page	int	Page number (default 1).	1

Returns:

Type	Description
EventExportResponse	EventExportResponse with events and total count.

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

def list_events(
    self,
    project: Optional[str] = None,
    filters: Optional[List[Union[EventFilter, Dict[str, Any]]]] = None,
    *,
    date_range: Optional[Dict[str, str]] = None,
    projections: Optional[List[str]] = None,
    limit: int = 1000,
    page: int = 1,
) -> EventExportResponse:
    """List events via export endpoint (backwards compatible alias).

    This is a backwards compatible alias for export(). Uses the Data Plane
    POST /events/export endpoint.

    Accepted keys:
        Project API key

    Args:
        project: Project name. Deprecated in v1.0 and will be removed in v2.0.
        filters: List of EventFilter objects or dicts.
        date_range: Optional date range filter.
        projections: Optional list of fields to include.
        limit: Maximum number of results (default 1000).
        page: Page number (default 1).

    Returns:
        EventExportResponse with events and total count.
    """
    return self.export(
        project=project,
        filters=filters,
        date_range=date_range,
        projections=projections,
        limit=limit,
        page=page,
    )

get_events

get_events(
    project: Optional[str] = None,
    filters: Optional[
        List[Union[EventFilter, Dict[str, Any]]]
    ] = None,
    *,
    date_range: Optional[Dict[str, str]] = None,
    projections: Optional[List[str]] = None,
    limit: int = 1000,
    page: int = 1
) -> EventExportResponse

Get events via export endpoint (backwards compatible alias).

This is a backwards compatible alias for export(). Uses the Data Plane POST /events/export endpoint.

Accepted keys

Project API key

Parameters:

Name	Type	Description	Default
project	Optional[str]	Project name. Deprecated in v1.0 and will be removed in v2.0.	None
filters	Optional[List[Union[EventFilter, Dict[str, Any]]]]	List of EventFilter objects or dicts.	None
date_range	Optional[Dict[str, str]]	Optional date range filter.	None
projections	Optional[List[str]]	Optional list of fields to include.	None
limit	int	Maximum number of results (default 1000).	1000
page	int	Page number (default 1).	1

Returns:

Type	Description
EventExportResponse	EventExportResponse with events and total count.

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

def get_events(
    self,
    project: Optional[str] = None,
    filters: Optional[List[Union[EventFilter, Dict[str, Any]]]] = None,
    *,
    date_range: Optional[Dict[str, str]] = None,
    projections: Optional[List[str]] = None,
    limit: int = 1000,
    page: int = 1,
) -> EventExportResponse:
    """Get events via export endpoint (backwards compatible alias).

    This is a backwards compatible alias for export(). Uses the Data Plane
    POST /events/export endpoint.

    Accepted keys:
        Project API key

    Args:
        project: Project name. Deprecated in v1.0 and will be removed in v2.0.
        filters: List of EventFilter objects or dicts.
        date_range: Optional date range filter.
        projections: Optional list of fields to include.
        limit: Maximum number of results (default 1000).
        page: Page number (default 1).

    Returns:
        EventExportResponse with events and total count.
    """
    return self.export(
        project=project,
        filters=filters,
        date_range=date_range,
        projections=projections,
        limit=limit,
        page=page,
    )

list_events_async async

list_events_async(
    project: Optional[str] = None,
    filters: Optional[
        List[Union[EventFilter, Dict[str, Any]]]
    ] = None,
    *,
    date_range: Optional[Dict[str, str]] = None,
    projections: Optional[List[str]] = None,
    limit: int = 1000,
    page: int = 1
) -> EventExportResponse

List events asynchronously (backwards compatible alias for export_async).

Accepted keys

Project API key

Parameters:

Name	Type	Description	Default
project	Optional[str]	Project name. Deprecated in v1.0 and will be removed in v2.0.	None

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

async def list_events_async(
    self,
    project: Optional[str] = None,
    filters: Optional[List[Union[EventFilter, Dict[str, Any]]]] = None,
    *,
    date_range: Optional[Dict[str, str]] = None,
    projections: Optional[List[str]] = None,
    limit: int = 1000,
    page: int = 1,
) -> EventExportResponse:
    """List events asynchronously (backwards compatible alias for export_async).

    Accepted keys:
        Project API key

    Args:
        project: Project name. Deprecated in v1.0 and will be removed in v2.0.
    """
    return await self.export_async(
        project=project,
        filters=filters,
        date_range=date_range,
        projections=projections,
        limit=limit,
        page=page,
    )

get_events_async async

get_events_async(
    project: Optional[str] = None,
    filters: Optional[
        List[Union[EventFilter, Dict[str, Any]]]
    ] = None,
    *,
    date_range: Optional[Dict[str, str]] = None,
    projections: Optional[List[str]] = None,
    limit: int = 1000,
    page: int = 1
) -> EventExportResponse

Get events asynchronously (backwards compatible alias for export_async).

Accepted keys

Project API key

Parameters:

Name	Type	Description	Default
project	Optional[str]	Project name. Deprecated in v1.0 and will be removed in v2.0.	None

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

async def get_events_async(
    self,
    project: Optional[str] = None,
    filters: Optional[List[Union[EventFilter, Dict[str, Any]]]] = None,
    *,
    date_range: Optional[Dict[str, str]] = None,
    projections: Optional[List[str]] = None,
    limit: int = 1000,
    page: int = 1,
) -> EventExportResponse:
    """Get events asynchronously (backwards compatible alias for export_async).

    Accepted keys:
        Project API key

    Args:
        project: Project name. Deprecated in v1.0 and will be removed in v2.0.
    """
    return await self.export_async(
        project=project,
        filters=filters,
        date_range=date_range,
        projections=projections,
        limit=limit,
        page=page,
    )

ExperimentsAPI

Bases: BaseAPI

Experiments API.

A method that takes project_id sends the request with the data plane API key when given it, naming the project in the path. Without it, the request acts on the project API key's project, is sent with that key, and issues a DeprecationWarning.

Each such public method warns in its own name and then calls its private core, which sends the request and never warns. SDK code that reaches the operation calls the core, so the warning reaches only a caller who can pass project_id.

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

class ExperimentsAPI(BaseAPI):
    """Experiments API.

    A method that takes ``project_id`` sends the request with the data plane
    API key when given it, naming the project in the path. Without it, the
    request acts on the project API key's project, is sent with that key, and
    issues a ``DeprecationWarning``.

    Each such public method warns in its own name and then calls its private
    core, which sends the request and never warns. SDK code that reaches the
    operation calls the core, so the warning reaches only a caller who can pass
    ``project_id``.
    """

    # Sync methods
    def get_schema(
        self,
        dateRange: Optional[Any] = None,
        evaluation_id: Optional[str] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetEventsSchemaResponse:
        """Get the events schema across a project's experiment runs, or of one run.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            dateRange: Filter by date range (string or dict with $gte/$lte).
            evaluation_id: Filter by evaluation/run ID: the schema covers that
                one run.
            project_id: The project whose runs the schema covers. Omitting it
                reads the project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "get_schema")
        return self._get_schema(dateRange, evaluation_id, project_id=project_id)

    def _get_schema(
        self,
        dateRange: Optional[Any] = None,
        evaluation_id: Optional[str] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetEventsSchemaResponse:
        if project_id is None:
            return events_svc.getEventsSchemaLegacy(
                self._api_config, dateRange=dateRange, evaluation_id=evaluation_id
            )
        if evaluation_id is not None:
            return experiments_svc.getRunSchema(
                self._data_plane_api_config,
                project_id=project_id,
                run_id=evaluation_id,
                dateRange=dateRange,
            )
        return experiments_svc.getRunsSchema(
            self._data_plane_api_config, project_id=project_id, dateRange=dateRange
        )

    def list_runs(
        self,
        dataset_id: Optional[str] = None,
        page: Optional[int] = None,
        limit: Optional[int] = None,
        run_ids: Optional[List[str]] = None,
        name: Optional[str] = None,
        status: Optional[str] = None,
        dateRange: Optional[Any] = None,
        sort_by: Optional[str] = None,
        sort_order: Optional[str] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetExperimentRunsResponse:
        """List experiment runs.

        When run_ids exceeds QUERY_BATCH_SIZE, requests are automatically
        batched to stay within URL length limits and the results are merged.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            dataset_id: Filter by dataset ID.
            page: Page number for pagination.
            limit: Number of results per page.
            run_ids: Filter by specific run IDs.
            name: Filter by run name.
            status: Filter by run status.
            dateRange: Filter by date range.
            sort_by: Sort by field.
            sort_order: Sort order (asc/desc).
            project_id: The project whose runs are listed. Omitting it lists the
                project API key's runs, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "list_runs")
        return self._list_runs(
            dataset_id=dataset_id,
            page=page,
            limit=limit,
            run_ids=run_ids,
            name=name,
            status=status,
            dateRange=dateRange,
            sort_by=sort_by,
            sort_order=sort_order,
            project_id=project_id,
        )

    def _batched_list_runs(
        self,
        send: Callable[..., GetExperimentRunsResponse],
        *,
        run_ids: List[str],
        **kwargs: Any,
    ) -> GetExperimentRunsResponse:
        """Fetch runs in batches through ``send`` and merge the responses."""
        # Pagination doesn't apply when batching by IDs — each batch must
        # return all matching runs for its chunk.
        kwargs.pop("page", None)
        kwargs.pop("limit", None)

        all_evaluations: List[Any] = []
        all_metrics: List[str] = []
        total = 0
        total_unfiltered = 0

        batches = _chunk_list(run_ids, QUERY_BATCH_SIZE)
        for i, batch in enumerate(batches):
            try:
                resp = send(run_ids=batch, **kwargs)
            except Exception:
                logger.warning(
                    "Batch %d/%d failed (%d run IDs)",
                    i + 1,
                    len(batches),
                    len(batch),
                )
                raise
            all_evaluations.extend(resp.evaluations)
            all_metrics.extend(resp.metrics)
            total += resp.pagination.total
            total_unfiltered += resp.pagination.total_unfiltered

        return GetExperimentRunsResponse.model_construct(
            evaluations=all_evaluations,
            metrics=list(dict.fromkeys(all_metrics)),  # deduplicate, preserve order
            pagination=Pagination(
                page=1,
                limit=total,
                total=total,
                total_unfiltered=total_unfiltered,
                total_pages=1,
                has_next=False,
                has_prev=False,
            ),
        )

    def get_run(
        self, run_id: str, *, project_id: Optional[str] = None
    ) -> GetExperimentRunResponse:
        """Get an experiment run by ID.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            run_id: The run to get.
            project_id: The project the run is read from. Omitting it reads
                from the project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "get_run")
        return self._get_run(run_id, project_id=project_id)

    def get_run_metrics(
        self,
        run_id: str,
        dateRange: Optional[str] = None,
        filters: Optional[Union[str, List[Dict[str, Any]]]] = None,
        *,
        project_id: Optional[str] = None,
    ) -> _GetExperimentRunMetricsResponse:
        """Get the event metrics of an experiment run.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            run_id: The run whose event metrics are read.
            dateRange: Date range filter as a JSON string.
            filters: Filters to apply, as a JSON string or a list of filter
                objects.
            project_id: The project that holds the run. Omitting it reads from
                the project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "get_run_metrics")
        return self._get_run_metrics(run_id, dateRange, filters, project_id=project_id)

    def create_run(
        self, request: PostExperimentRunRequest, *, project_id: Optional[str] = None
    ) -> PostExperimentRunResponse:
        """Create an experiment run.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            request: The run to create.
            project_id: The project the run is created in. Omitting it creates
                it in the project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "create_run")
        return self._create_run(request, project_id=project_id)

    def update_run(
        self,
        run_id: str,
        request: PutExperimentRunRequest,
        *,
        project_id: Optional[str] = None,
    ) -> PutExperimentRunResponse:
        """Update an experiment run.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            run_id: The run to update.
            request: The fields to update.
            project_id: The project that holds the run. Omitting it updates the
                run in the project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "update_run")
        return self._update_run(run_id, request, project_id=project_id)

    def delete_run(
        self, run_id: str, *, project_id: Optional[str] = None
    ) -> DeleteExperimentRunResponse:
        """Delete an experiment run.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            run_id: The run to delete.
            project_id: The project that holds the run. Omitting it deletes the
                run in the project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "delete_run")
        return self._delete_run(run_id, project_id=project_id)

    # Async methods
    async def get_schema_async(
        self,
        dateRange: Optional[Any] = None,
        evaluation_id: Optional[str] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetEventsSchemaResponse:
        """Get the events schema across a project's experiment runs, or of one run, asynchronously.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            dateRange: Filter by date range (string or dict with $gte/$lte).
            evaluation_id: Filter by evaluation/run ID: the schema covers that
                one run.
            project_id: The project whose runs the schema covers. Omitting it
                reads the project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "get_schema_async")
        return await self._get_schema_async(
            dateRange, evaluation_id, project_id=project_id
        )

    async def _get_schema_async(
        self,
        dateRange: Optional[Any] = None,
        evaluation_id: Optional[str] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetEventsSchemaResponse:
        if project_id is None:
            return await events_svc_async.getEventsSchemaLegacy(
                self._api_config, dateRange=dateRange, evaluation_id=evaluation_id
            )
        if evaluation_id is not None:
            return await experiments_svc_async.getRunSchema(
                self._data_plane_api_config,
                project_id=project_id,
                run_id=evaluation_id,
                dateRange=dateRange,
            )
        return await experiments_svc_async.getRunsSchema(
            self._data_plane_api_config, project_id=project_id, dateRange=dateRange
        )

    async def list_runs_async(
        self,
        dataset_id: Optional[str] = None,
        page: Optional[int] = None,
        limit: Optional[int] = None,
        run_ids: Optional[List[str]] = None,
        name: Optional[str] = None,
        status: Optional[str] = None,
        dateRange: Optional[Any] = None,
        sort_by: Optional[str] = None,
        sort_order: Optional[str] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetExperimentRunsResponse:
        """List experiment runs asynchronously. See ``list_runs()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "list_runs_async")
        return await self._list_runs_async(
            dataset_id=dataset_id,
            page=page,
            limit=limit,
            run_ids=run_ids,
            name=name,
            status=status,
            dateRange=dateRange,
            sort_by=sort_by,
            sort_order=sort_order,
            project_id=project_id,
        )

    async def _batched_list_runs_async(
        self,
        send: Callable[..., Awaitable[GetExperimentRunsResponse]],
        *,
        run_ids: List[str],
        **kwargs: Any,
    ) -> GetExperimentRunsResponse:
        """Fetch runs in batches (async, parallel) through ``send`` and merge the responses."""
        # Strip pagination params — each batch fetches its full slice.
        kwargs.pop("page", None)
        kwargs.pop("limit", None)

        batches = _chunk_list(run_ids, QUERY_BATCH_SIZE)
        resps = await asyncio.gather(
            *(send(run_ids=batch, **kwargs) for batch in batches)
        )

        all_evaluations: List[Any] = []
        all_metrics: List[str] = []
        total = 0
        total_unfiltered = 0
        for resp in resps:
            all_evaluations.extend(resp.evaluations)
            all_metrics.extend(resp.metrics)
            total += resp.pagination.total
            total_unfiltered += resp.pagination.total_unfiltered

        return GetExperimentRunsResponse.model_construct(
            evaluations=all_evaluations,
            metrics=list(dict.fromkeys(all_metrics)),  # deduplicate, preserve order
            pagination=Pagination(
                page=1,
                limit=total,
                total=total,
                total_unfiltered=total_unfiltered,
                total_pages=1,
                has_next=False,
                has_prev=False,
            ),
        )

    async def get_run_metrics_async(
        self,
        run_id: str,
        dateRange: Optional[str] = None,
        filters: Optional[Union[str, List[Dict[str, Any]]]] = None,
        *,
        project_id: Optional[str] = None,
    ) -> _GetExperimentRunMetricsResponse:
        """Get the event metrics of an experiment run asynchronously.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            run_id: The run whose event metrics are read.
            dateRange: Date range filter as a JSON string.
            filters: Filters to apply, as a JSON string or a list of filter
                objects.
            project_id: The project that holds the run. Omitting it reads from
                the project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "get_run_metrics_async")
        return await self._get_run_metrics_async(
            run_id, dateRange, filters, project_id=project_id
        )

    async def get_run_async(
        self, run_id: str, *, project_id: Optional[str] = None
    ) -> GetExperimentRunResponse:
        """Get an experiment run by ID asynchronously. See ``get_run()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "get_run_async")
        return await self._get_run_async(run_id, project_id=project_id)

    async def create_run_async(
        self, request: PostExperimentRunRequest, *, project_id: Optional[str] = None
    ) -> PostExperimentRunResponse:
        """Create an experiment run asynchronously. See ``create_run()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "create_run_async")
        return await self._create_run_async(request, project_id=project_id)

    async def update_run_async(
        self,
        run_id: str,
        request: PutExperimentRunRequest,
        *,
        project_id: Optional[str] = None,
    ) -> PutExperimentRunResponse:
        """Update an experiment run asynchronously. See ``update_run()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "update_run_async")
        return await self._update_run_async(run_id, request, project_id=project_id)

    async def delete_run_async(
        self, run_id: str, *, project_id: Optional[str] = None
    ) -> DeleteExperimentRunResponse:
        """Delete an experiment run asynchronously. See ``delete_run()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "delete_run_async")
        return await self._delete_run_async(run_id, project_id=project_id)

    def get_result(
        self,
        run_id: str,
        aggregate_function: Optional[str] = None,
        filters: Optional[Any] = None,
        *,
        project_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Get experiment run result.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            run_id: The experiment run ID.
            aggregate_function: Aggregation function to apply. With
                ``project_id`` it must be one of ``average``, ``min``, ``max``,
                ``median``, ``p90``, ``p95``, ``p99``, ``sum`` or ``count``;
                any other value answers 400. Without ``project_id`` an
                unknown value falls back to ``average``.
            filters: Optional filters to apply.
            project_id: The project that holds the run. Omitting it reads the
                project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "get_result")
        return self._get_result(
            run_id, aggregate_function, filters, project_id=project_id
        )

    def _get_result(
        self,
        run_id: str,
        aggregate_function: Optional[str] = None,
        filters: Optional[Any] = None,
        *,
        project_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        if project_id is None:
            result = experiments_svc.getExperimentResultLegacy(
                self._api_config,
                run_id=run_id,
                aggregate_function=aggregate_function,
                filters=filters,
            )
        else:
            result = experiments_svc.getExperimentSummary(
                self._data_plane_api_config,
                project_id=project_id,
                run_id=run_id,
                aggregate_function=aggregate_function,
                filters=filters,
            )
        # GetExperimentRunResultResponse is a pass-through dict model
        return result.model_dump() if hasattr(result, "model_dump") else dict(result)

    def compare_runs(
        self,
        new_run_id: str,
        old_run_id: str,
        aggregate_function: Optional[str] = None,
        filters: Optional[Any] = None,
        *,
        project_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Compare two experiment runs.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            new_run_id: The new run ID to compare.
            old_run_id: The old run ID to compare against.
            aggregate_function: Aggregation function to apply.
            filters: Optional filters to apply.
            project_id: The project that holds both runs. Omitting it compares
                runs in the project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "compare_runs")
        return self._compare_runs(
            new_run_id,
            old_run_id,
            aggregate_function,
            filters,
            project_id=project_id,
        )

    async def get_result_async(
        self,
        run_id: str,
        aggregate_function: Optional[str] = None,
        filters: Optional[Any] = None,
        *,
        project_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Get experiment run result asynchronously. See ``get_result()``.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "get_result_async")
        return await self._get_result_async(
            run_id, aggregate_function, filters, project_id=project_id
        )

    async def _get_result_async(
        self,
        run_id: str,
        aggregate_function: Optional[str] = None,
        filters: Optional[Any] = None,
        *,
        project_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        if project_id is None:
            result = await experiments_svc_async.getExperimentResultLegacy(
                self._api_config,
                run_id=run_id,
                aggregate_function=aggregate_function,
                filters=filters,
            )
        else:
            result = await experiments_svc_async.getExperimentSummary(
                self._data_plane_api_config,
                project_id=project_id,
                run_id=run_id,
                aggregate_function=aggregate_function,
                filters=filters,
            )
        return result.model_dump() if hasattr(result, "model_dump") else dict(result)

    async def compare_runs_async(
        self,
        new_run_id: str,
        old_run_id: str,
        aggregate_function: Optional[str] = None,
        filters: Optional[Any] = None,
        *,
        project_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Compare two experiment runs asynchronously.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            new_run_id: The new run ID to compare.
            old_run_id: The old run ID to compare against.
            aggregate_function: Aggregation function to apply.
            filters: Optional filters to apply.
            project_id: The project that holds both runs. Omitting it compares
                runs in the project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "compare_runs_async")
        return await self._compare_runs_async(
            new_run_id,
            old_run_id,
            aggregate_function,
            filters,
            project_id=project_id,
        )

    # Aliases for backwards compatibility (evaluations naming)
    def get_run_result(
        self,
        run_id: str,
        aggregate_function: Optional[str] = None,
        filters: Optional[Any] = None,
        *,
        project_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Get experiment run result (alias for get_result).

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "get_run_result")
        return self._get_result(
            run_id, aggregate_function, filters, project_id=project_id
        )

    def compare_run_events(
        self,
        new_run_id: str,
        old_run_id: str,
        event_name: Optional[str] = None,
        event_type: Optional[str] = None,
        filter: Optional[Any] = None,
        limit: Optional[int] = None,
        page: Optional[int] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetExperimentCompareEventsResponse:
        """Compare events between two experiment runs.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            new_run_id: The new run ID to compare.
            old_run_id: The old run ID to compare against.
            event_name: Filter by event name.
            event_type: Filter by event type.
            filter: Additional filter criteria.
            limit: Maximum number of results.
            page: Page number for pagination.
            project_id: The project that holds both runs. Omitting it reads the
                project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "compare_run_events")
        return self._compare_run_events(
            new_run_id,
            old_run_id,
            event_name,
            event_type,
            filter,
            limit,
            page,
            project_id=project_id,
        )

    def _compare_run_events(
        self,
        new_run_id: str,
        old_run_id: str,
        event_name: Optional[str] = None,
        event_type: Optional[str] = None,
        filter: Optional[Any] = None,
        limit: Optional[int] = None,
        page: Optional[int] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetExperimentCompareEventsResponse:
        """The request ``compare_run_events()`` sends."""
        if project_id is None:
            return experiments_svc.getExperimentCompareEventsQueryIdLegacy(
                self._api_config,
                run_id_1=new_run_id,
                run_id_2=old_run_id,
                event_name=event_name,
                event_type=event_type,
                filter=filter,
                limit=limit,
                page=page,
            )
        return experiments_svc.getExperimentCompareEvents(
            self._data_plane_api_config,
            project_id=project_id,
            new_run_id=new_run_id,
            old_run_id=old_run_id,
            event_name=event_name,
            event_type=event_type,
            filter=filter,
            limit=limit,
            page=page,
        )

    async def compare_run_events_async(
        self,
        new_run_id: str,
        old_run_id: str,
        event_name: Optional[str] = None,
        event_type: Optional[str] = None,
        filter: Optional[Any] = None,
        limit: Optional[int] = None,
        page: Optional[int] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetExperimentCompareEventsResponse:
        """Compare events between two experiment runs asynchronously.

        Accepted keys:
            Data plane fine-grained API key with ``project_id``; Project API key
            without ``project_id`` (deprecated)

        Args:
            new_run_id: The new run ID to compare.
            old_run_id: The old run ID to compare against.
            event_name: Filter by event name.
            event_type: Filter by event type.
            filter: Additional filter criteria.
            limit: Maximum number of results.
            page: Page number for pagination.
            project_id: The project that holds both runs. Omitting it reads the
                project API key's project, which is deprecated.
        """
        if project_id is None:
            _warn_call_without_project_id("experiments", "compare_run_events_async")
        return await self._compare_run_events_async(
            new_run_id,
            old_run_id,
            event_name,
            event_type,
            filter,
            limit,
            page,
            project_id=project_id,
        )

    async def _compare_run_events_async(
        self,
        new_run_id: str,
        old_run_id: str,
        event_name: Optional[str] = None,
        event_type: Optional[str] = None,
        filter: Optional[Any] = None,
        limit: Optional[int] = None,
        page: Optional[int] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetExperimentCompareEventsResponse:
        """The request ``compare_run_events_async()`` sends."""
        if project_id is None:
            return await experiments_svc_async.getExperimentCompareEventsQueryIdLegacy(
                self._api_config,
                run_id_1=new_run_id,
                run_id_2=old_run_id,
                event_name=event_name,
                event_type=event_type,
                filter=filter,
                limit=limit,
                page=page,
            )
        return await experiments_svc_async.getExperimentCompareEvents(
            self._data_plane_api_config,
            project_id=project_id,
            new_run_id=new_run_id,
            old_run_id=old_run_id,
            event_name=event_name,
            event_type=event_type,
            filter=filter,
            limit=limit,
            page=page,
        )

    def _get_run_metrics(
        self,
        run_id: str,
        dateRange: Optional[str] = None,
        filters: Optional[Union[str, List[Dict[str, Any]]]] = None,
        *,
        project_id: Optional[str] = None,
    ) -> _GetExperimentRunMetricsResponse:
        """The request ``get_run_metrics()`` sends."""
        if project_id is None:
            return experiments_svc.getExperimentRunMetricsLegacy(
                self._api_config, run_id=run_id, dateRange=dateRange, filters=filters
            )
        return experiments_svc.getExperimentRunMetrics(
            self._data_plane_api_config,
            project_id=project_id,
            run_id=run_id,
            dateRange=dateRange,
            filters=filters,
        )

    def _get_run(
        self, run_id: str, *, project_id: Optional[str] = None
    ) -> GetExperimentRunResponse:
        """The request ``get_run()`` sends."""
        if project_id is None:
            return experiments_svc.getRunLegacy(self._api_config, run_id=run_id)
        return experiments_svc.getRun(
            self._data_plane_api_config, project_id=project_id, run_id=run_id
        )

    async def _get_run_metrics_async(
        self,
        run_id: str,
        dateRange: Optional[str] = None,
        filters: Optional[Union[str, List[Dict[str, Any]]]] = None,
        *,
        project_id: Optional[str] = None,
    ) -> _GetExperimentRunMetricsResponse:
        """The request ``get_run_metrics_async()`` sends."""
        if project_id is None:
            return await experiments_svc_async.getExperimentRunMetricsLegacy(
                self._api_config, run_id=run_id, dateRange=dateRange, filters=filters
            )
        return await experiments_svc_async.getExperimentRunMetrics(
            self._data_plane_api_config,
            project_id=project_id,
            run_id=run_id,
            dateRange=dateRange,
            filters=filters,
        )

    async def _get_run_async(
        self, run_id: str, *, project_id: Optional[str] = None
    ) -> GetExperimentRunResponse:
        """The request ``get_run_async()`` sends."""
        if project_id is None:
            return await experiments_svc_async.getRunLegacy(
                self._api_config, run_id=run_id
            )
        return await experiments_svc_async.getRun(
            self._data_plane_api_config, project_id=project_id, run_id=run_id
        )

    def _compare_runs(
        self,
        new_run_id: str,
        old_run_id: str,
        aggregate_function: Optional[str] = None,
        filters: Optional[Any] = None,
        *,
        project_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        """The request ``compare_runs()`` sends."""
        if project_id is None:
            result = experiments_svc.getExperimentComparisonCompareWithLegacy(
                self._api_config,
                new_run_id=new_run_id,
                old_run_id=old_run_id,
                aggregate_function=aggregate_function,
                filters=filters,
            )
        else:
            result = experiments_svc.getExperimentComparison(
                self._data_plane_api_config,
                project_id=project_id,
                new_run_id=new_run_id,
                old_run_id=old_run_id,
                aggregate_function=aggregate_function,
                filters=filters,
            )
        # GetExperimentRunCompareResponse is a pass-through dict model
        return result.model_dump() if hasattr(result, "model_dump") else dict(result)

    async def _compare_runs_async(
        self,
        new_run_id: str,
        old_run_id: str,
        aggregate_function: Optional[str] = None,
        filters: Optional[Any] = None,
        *,
        project_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        """The request ``compare_runs_async()`` sends."""
        if project_id is None:
            result = (
                await experiments_svc_async.getExperimentComparisonCompareWithLegacy(
                    self._api_config,
                    new_run_id=new_run_id,
                    old_run_id=old_run_id,
                    aggregate_function=aggregate_function,
                    filters=filters,
                )
            )
        else:
            result = await experiments_svc_async.getExperimentComparison(
                self._data_plane_api_config,
                project_id=project_id,
                new_run_id=new_run_id,
                old_run_id=old_run_id,
                aggregate_function=aggregate_function,
                filters=filters,
            )
        return result.model_dump() if hasattr(result, "model_dump") else dict(result)

    def _create_run(
        self, request: PostExperimentRunRequest, *, project_id: Optional[str] = None
    ) -> PostExperimentRunResponse:
        """The request ``create_run()`` sends."""
        if project_id is None:
            return experiments_svc.createRunLegacy(self._api_config, data=request)
        return experiments_svc.createRun(
            self._data_plane_api_config, project_id=project_id, data=request
        )

    async def _create_run_async(
        self, request: PostExperimentRunRequest, *, project_id: Optional[str] = None
    ) -> PostExperimentRunResponse:
        """The request ``create_run_async()`` sends."""
        if project_id is None:
            return await experiments_svc_async.createRunLegacy(
                self._api_config, data=request
            )
        return await experiments_svc_async.createRun(
            self._data_plane_api_config, project_id=project_id, data=request
        )

    def _list_runs(
        self,
        dataset_id: Optional[str] = None,
        page: Optional[int] = None,
        limit: Optional[int] = None,
        run_ids: Optional[List[str]] = None,
        name: Optional[str] = None,
        status: Optional[str] = None,
        dateRange: Optional[Any] = None,
        sort_by: Optional[str] = None,
        sort_order: Optional[str] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetExperimentRunsResponse:
        """The request ``list_runs()`` sends."""
        if project_id is None:
            api_config = self._api_config

            def send(**query: Any) -> GetExperimentRunsResponse:
                return experiments_svc.getRunsLegacy(api_config, **query)

        else:
            api_config = self._data_plane_api_config
            scoped_project_id = project_id

            def send(**query: Any) -> GetExperimentRunsResponse:
                return experiments_svc.getRuns(
                    api_config, project_id=scoped_project_id, **query
                )

        # Batch if the list is large enough to risk exceeding URL length limits.
        if run_ids and len(run_ids) > QUERY_BATCH_SIZE:
            return self._batched_list_runs(
                send,
                dataset_id=dataset_id,
                page=page,
                limit=limit,
                run_ids=run_ids,
                name=name,
                status=status,
                dateRange=dateRange,
                sort_by=sort_by,
                sort_order=sort_order,
            )
        return send(
            dataset_id=dataset_id,
            page=page,
            limit=limit,
            run_ids=run_ids,
            name=name,
            status=status,
            dateRange=dateRange,
            sort_by=sort_by,
            sort_order=sort_order,
        )

    async def _list_runs_async(
        self,
        dataset_id: Optional[str] = None,
        page: Optional[int] = None,
        limit: Optional[int] = None,
        run_ids: Optional[List[str]] = None,
        name: Optional[str] = None,
        status: Optional[str] = None,
        dateRange: Optional[Any] = None,
        sort_by: Optional[str] = None,
        sort_order: Optional[str] = None,
        *,
        project_id: Optional[str] = None,
    ) -> GetExperimentRunsResponse:
        """The request ``list_runs_async()`` sends."""
        if project_id is None:
            api_config = self._api_config

            async def send(**query: Any) -> GetExperimentRunsResponse:
                return await experiments_svc_async.getRunsLegacy(api_config, **query)

        else:
            api_config = self._data_plane_api_config
            scoped_project_id = project_id

            async def send(**query: Any) -> GetExperimentRunsResponse:
                return await experiments_svc_async.getRuns(
                    api_config, project_id=scoped_project_id, **query
                )

        # Batch if the list is large enough to risk exceeding URL length limits.
        if run_ids and len(run_ids) > QUERY_BATCH_SIZE:
            return await self._batched_list_runs_async(
                send,
                dataset_id=dataset_id,
                page=page,
                limit=limit,
                run_ids=run_ids,
                name=name,
                status=status,
                dateRange=dateRange,
                sort_by=sort_by,
                sort_order=sort_order,
            )
        return await send(
            dataset_id=dataset_id,
            page=page,
            limit=limit,
            run_ids=run_ids,
            name=name,
            status=status,
            dateRange=dateRange,
            sort_by=sort_by,
            sort_order=sort_order,
        )

    def _update_run(
        self,
        run_id: str,
        request: PutExperimentRunRequest,
        *,
        project_id: Optional[str] = None,
    ) -> PutExperimentRunResponse:
        """The request ``update_run()`` sends."""
        if project_id is None:
            return experiments_svc.updateRunLegacy(
                self._api_config, run_id=run_id, data=request
            )
        return experiments_svc.updateRun(
            self._data_plane_api_config,
            project_id=project_id,
            run_id=run_id,
            data=request,
        )

    async def _update_run_async(
        self,
        run_id: str,
        request: PutExperimentRunRequest,
        *,
        project_id: Optional[str] = None,
    ) -> PutExperimentRunResponse:
        """The request ``update_run_async()`` sends."""
        if project_id is None:
            return await experiments_svc_async.updateRunLegacy(
                self._api_config, run_id=run_id, data=request
            )
        return await experiments_svc_async.updateRun(
            self._data_plane_api_config,
            project_id=project_id,
            run_id=run_id,
            data=request,
        )

    def _delete_run(
        self, run_id: str, *, project_id: Optional[str] = None
    ) -> DeleteExperimentRunResponse:
        """The request ``delete_run()`` sends."""
        if project_id is None:
            return experiments_svc.deleteRunLegacy(self._api_config, run_id=run_id)
        return experiments_svc.deleteRun(
            self._data_plane_api_config, project_id=project_id, run_id=run_id
        )

    async def _delete_run_async(
        self, run_id: str, *, project_id: Optional[str] = None
    ) -> DeleteExperimentRunResponse:
        """The request ``delete_run_async()`` sends."""
        if project_id is None:
            return await experiments_svc_async.deleteRunLegacy(
                self._api_config, run_id=run_id
            )
        return await experiments_svc_async.deleteRun(
            self._data_plane_api_config, project_id=project_id, run_id=run_id
        )

get_schema

get_schema(
    dateRange: Optional[Any] = None,
    evaluation_id: Optional[str] = None,
    *,
    project_id: Optional[str] = None
) -> GetEventsSchemaResponse

Get the events schema across a project's experiment runs, or of one run.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
dateRange	Optional[Any]	Filter by date range (string or dict with $gte/$lte).	None
evaluation_id	Optional[str]	Filter by evaluation/run ID: the schema covers that one run.	None
project_id	Optional[str]	The project whose runs the schema covers. Omitting it reads the project API key's project, which is deprecated.	None

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

def get_schema(
    self,
    dateRange: Optional[Any] = None,
    evaluation_id: Optional[str] = None,
    *,
    project_id: Optional[str] = None,
) -> GetEventsSchemaResponse:
    """Get the events schema across a project's experiment runs, or of one run.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        dateRange: Filter by date range (string or dict with $gte/$lte).
        evaluation_id: Filter by evaluation/run ID: the schema covers that
            one run.
        project_id: The project whose runs the schema covers. Omitting it
            reads the project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "get_schema")
    return self._get_schema(dateRange, evaluation_id, project_id=project_id)

list_runs

list_runs(
    dataset_id: Optional[str] = None,
    page: Optional[int] = None,
    limit: Optional[int] = None,
    run_ids: Optional[List[str]] = None,
    name: Optional[str] = None,
    status: Optional[str] = None,
    dateRange: Optional[Any] = None,
    sort_by: Optional[str] = None,
    sort_order: Optional[str] = None,
    *,
    project_id: Optional[str] = None
) -> GetExperimentRunsResponse

List experiment runs.

When run_ids exceeds QUERY_BATCH_SIZE, requests are automatically batched to stay within URL length limits and the results are merged.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
dataset_id	Optional[str]	Filter by dataset ID.	None
page	Optional[int]	Page number for pagination.	None
limit	Optional[int]	Number of results per page.	None
run_ids	Optional[List[str]]	Filter by specific run IDs.	None
name	Optional[str]	Filter by run name.	None
status	Optional[str]	Filter by run status.	None
dateRange	Optional[Any]	Filter by date range.	None
sort_by	Optional[str]	Sort by field.	None
sort_order	Optional[str]	Sort order (asc/desc).	None
project_id	Optional[str]	The project whose runs are listed. Omitting it lists the project API key's runs, which is deprecated.	None

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

def list_runs(
    self,
    dataset_id: Optional[str] = None,
    page: Optional[int] = None,
    limit: Optional[int] = None,
    run_ids: Optional[List[str]] = None,
    name: Optional[str] = None,
    status: Optional[str] = None,
    dateRange: Optional[Any] = None,
    sort_by: Optional[str] = None,
    sort_order: Optional[str] = None,
    *,
    project_id: Optional[str] = None,
) -> GetExperimentRunsResponse:
    """List experiment runs.

    When run_ids exceeds QUERY_BATCH_SIZE, requests are automatically
    batched to stay within URL length limits and the results are merged.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        dataset_id: Filter by dataset ID.
        page: Page number for pagination.
        limit: Number of results per page.
        run_ids: Filter by specific run IDs.
        name: Filter by run name.
        status: Filter by run status.
        dateRange: Filter by date range.
        sort_by: Sort by field.
        sort_order: Sort order (asc/desc).
        project_id: The project whose runs are listed. Omitting it lists the
            project API key's runs, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "list_runs")
    return self._list_runs(
        dataset_id=dataset_id,
        page=page,
        limit=limit,
        run_ids=run_ids,
        name=name,
        status=status,
        dateRange=dateRange,
        sort_by=sort_by,
        sort_order=sort_order,
        project_id=project_id,
    )

get_run

get_run(
    run_id: str, *, project_id: Optional[str] = None
) -> GetExperimentRunResponse

Get an experiment run by ID.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
run_id	str	The run to get.	required
project_id	Optional[str]	The project the run is read from. Omitting it reads from the project API key's project, which is deprecated.	None

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

def get_run(
    self, run_id: str, *, project_id: Optional[str] = None
) -> GetExperimentRunResponse:
    """Get an experiment run by ID.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        run_id: The run to get.
        project_id: The project the run is read from. Omitting it reads
            from the project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "get_run")
    return self._get_run(run_id, project_id=project_id)

get_run_metrics

get_run_metrics(
    run_id: str,
    dateRange: Optional[str] = None,
    filters: Optional[
        Union[str, List[Dict[str, Any]]]
    ] = None,
    *,
    project_id: Optional[str] = None
) -> GetExperimentRunMetricsResponse

Get the event metrics of an experiment run.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
run_id	str	The run whose event metrics are read.	required
dateRange	Optional[str]	Date range filter as a JSON string.	None
filters	Optional[Union[str, List[Dict[str, Any]]]]	Filters to apply, as a JSON string or a list of filter objects.	None
project_id	Optional[str]	The project that holds the run. Omitting it reads from the project API key's project, which is deprecated.	None

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

def get_run_metrics(
    self,
    run_id: str,
    dateRange: Optional[str] = None,
    filters: Optional[Union[str, List[Dict[str, Any]]]] = None,
    *,
    project_id: Optional[str] = None,
) -> _GetExperimentRunMetricsResponse:
    """Get the event metrics of an experiment run.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        run_id: The run whose event metrics are read.
        dateRange: Date range filter as a JSON string.
        filters: Filters to apply, as a JSON string or a list of filter
            objects.
        project_id: The project that holds the run. Omitting it reads from
            the project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "get_run_metrics")
    return self._get_run_metrics(run_id, dateRange, filters, project_id=project_id)

create_run

create_run(
    request: PostExperimentRunRequest,
    *,
    project_id: Optional[str] = None
) -> PostExperimentRunResponse

Create an experiment run.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
request	PostExperimentRunRequest	The run to create.	required
project_id	Optional[str]	The project the run is created in. Omitting it creates it in the project API key's project, which is deprecated.	None

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

def create_run(
    self, request: PostExperimentRunRequest, *, project_id: Optional[str] = None
) -> PostExperimentRunResponse:
    """Create an experiment run.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        request: The run to create.
        project_id: The project the run is created in. Omitting it creates
            it in the project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "create_run")
    return self._create_run(request, project_id=project_id)

update_run

update_run(
    run_id: str,
    request: PutExperimentRunRequest,
    *,
    project_id: Optional[str] = None
) -> PutExperimentRunResponse

Update an experiment run.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
run_id	str	The run to update.	required
request	PutExperimentRunRequest	The fields to update.	required
project_id	Optional[str]	The project that holds the run. Omitting it updates the run in the project API key's project, which is deprecated.	None

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

def update_run(
    self,
    run_id: str,
    request: PutExperimentRunRequest,
    *,
    project_id: Optional[str] = None,
) -> PutExperimentRunResponse:
    """Update an experiment run.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        run_id: The run to update.
        request: The fields to update.
        project_id: The project that holds the run. Omitting it updates the
            run in the project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "update_run")
    return self._update_run(run_id, request, project_id=project_id)

delete_run

delete_run(
    run_id: str, *, project_id: Optional[str] = None
) -> DeleteExperimentRunResponse

Delete an experiment run.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
run_id	str	The run to delete.	required
project_id	Optional[str]	The project that holds the run. Omitting it deletes the run in the project API key's project, which is deprecated.	None

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

def delete_run(
    self, run_id: str, *, project_id: Optional[str] = None
) -> DeleteExperimentRunResponse:
    """Delete an experiment run.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        run_id: The run to delete.
        project_id: The project that holds the run. Omitting it deletes the
            run in the project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "delete_run")
    return self._delete_run(run_id, project_id=project_id)

get_schema_async async

get_schema_async(
    dateRange: Optional[Any] = None,
    evaluation_id: Optional[str] = None,
    *,
    project_id: Optional[str] = None
) -> GetEventsSchemaResponse

Get the events schema across a project's experiment runs, or of one run, asynchronously.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
dateRange	Optional[Any]	Filter by date range (string or dict with $gte/$lte).	None
evaluation_id	Optional[str]	Filter by evaluation/run ID: the schema covers that one run.	None
project_id	Optional[str]	The project whose runs the schema covers. Omitting it reads the project API key's project, which is deprecated.	None

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

async def get_schema_async(
    self,
    dateRange: Optional[Any] = None,
    evaluation_id: Optional[str] = None,
    *,
    project_id: Optional[str] = None,
) -> GetEventsSchemaResponse:
    """Get the events schema across a project's experiment runs, or of one run, asynchronously.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        dateRange: Filter by date range (string or dict with $gte/$lte).
        evaluation_id: Filter by evaluation/run ID: the schema covers that
            one run.
        project_id: The project whose runs the schema covers. Omitting it
            reads the project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "get_schema_async")
    return await self._get_schema_async(
        dateRange, evaluation_id, project_id=project_id
    )

list_runs_async async

list_runs_async(
    dataset_id: Optional[str] = None,
    page: Optional[int] = None,
    limit: Optional[int] = None,
    run_ids: Optional[List[str]] = None,
    name: Optional[str] = None,
    status: Optional[str] = None,
    dateRange: Optional[Any] = None,
    sort_by: Optional[str] = None,
    sort_order: Optional[str] = None,
    *,
    project_id: Optional[str] = None
) -> GetExperimentRunsResponse

List experiment runs asynchronously. See list_runs().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def list_runs_async(
    self,
    dataset_id: Optional[str] = None,
    page: Optional[int] = None,
    limit: Optional[int] = None,
    run_ids: Optional[List[str]] = None,
    name: Optional[str] = None,
    status: Optional[str] = None,
    dateRange: Optional[Any] = None,
    sort_by: Optional[str] = None,
    sort_order: Optional[str] = None,
    *,
    project_id: Optional[str] = None,
) -> GetExperimentRunsResponse:
    """List experiment runs asynchronously. See ``list_runs()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "list_runs_async")
    return await self._list_runs_async(
        dataset_id=dataset_id,
        page=page,
        limit=limit,
        run_ids=run_ids,
        name=name,
        status=status,
        dateRange=dateRange,
        sort_by=sort_by,
        sort_order=sort_order,
        project_id=project_id,
    )

get_run_metrics_async async

get_run_metrics_async(
    run_id: str,
    dateRange: Optional[str] = None,
    filters: Optional[
        Union[str, List[Dict[str, Any]]]
    ] = None,
    *,
    project_id: Optional[str] = None
) -> GetExperimentRunMetricsResponse

Get the event metrics of an experiment run asynchronously.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
run_id	str	The run whose event metrics are read.	required
dateRange	Optional[str]	Date range filter as a JSON string.	None
filters	Optional[Union[str, List[Dict[str, Any]]]]	Filters to apply, as a JSON string or a list of filter objects.	None
project_id	Optional[str]	The project that holds the run. Omitting it reads from the project API key's project, which is deprecated.	None

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

async def get_run_metrics_async(
    self,
    run_id: str,
    dateRange: Optional[str] = None,
    filters: Optional[Union[str, List[Dict[str, Any]]]] = None,
    *,
    project_id: Optional[str] = None,
) -> _GetExperimentRunMetricsResponse:
    """Get the event metrics of an experiment run asynchronously.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        run_id: The run whose event metrics are read.
        dateRange: Date range filter as a JSON string.
        filters: Filters to apply, as a JSON string or a list of filter
            objects.
        project_id: The project that holds the run. Omitting it reads from
            the project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "get_run_metrics_async")
    return await self._get_run_metrics_async(
        run_id, dateRange, filters, project_id=project_id
    )

get_run_async async

get_run_async(
    run_id: str, *, project_id: Optional[str] = None
) -> GetExperimentRunResponse

Get an experiment run by ID asynchronously. See get_run().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def get_run_async(
    self, run_id: str, *, project_id: Optional[str] = None
) -> GetExperimentRunResponse:
    """Get an experiment run by ID asynchronously. See ``get_run()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "get_run_async")
    return await self._get_run_async(run_id, project_id=project_id)

create_run_async async

create_run_async(
    request: PostExperimentRunRequest,
    *,
    project_id: Optional[str] = None
) -> PostExperimentRunResponse

Create an experiment run asynchronously. See create_run().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def create_run_async(
    self, request: PostExperimentRunRequest, *, project_id: Optional[str] = None
) -> PostExperimentRunResponse:
    """Create an experiment run asynchronously. See ``create_run()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "create_run_async")
    return await self._create_run_async(request, project_id=project_id)

update_run_async async

update_run_async(
    run_id: str,
    request: PutExperimentRunRequest,
    *,
    project_id: Optional[str] = None
) -> PutExperimentRunResponse

Update an experiment run asynchronously. See update_run().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def update_run_async(
    self,
    run_id: str,
    request: PutExperimentRunRequest,
    *,
    project_id: Optional[str] = None,
) -> PutExperimentRunResponse:
    """Update an experiment run asynchronously. See ``update_run()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "update_run_async")
    return await self._update_run_async(run_id, request, project_id=project_id)

delete_run_async async

delete_run_async(
    run_id: str, *, project_id: Optional[str] = None
) -> DeleteExperimentRunResponse

Delete an experiment run asynchronously. See delete_run().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def delete_run_async(
    self, run_id: str, *, project_id: Optional[str] = None
) -> DeleteExperimentRunResponse:
    """Delete an experiment run asynchronously. See ``delete_run()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "delete_run_async")
    return await self._delete_run_async(run_id, project_id=project_id)

get_result

get_result(
    run_id: str,
    aggregate_function: Optional[str] = None,
    filters: Optional[Any] = None,
    *,
    project_id: Optional[str] = None
) -> Dict[str, Any]

Get experiment run result.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
run_id	str	The experiment run ID.	required
aggregate_function	Optional[str]	Aggregation function to apply. With project_id it must be one of average, min, max, median, p90, p95, p99, sum or count; any other value answers 400. Without project_id an unknown value falls back to average.	None
filters	Optional[Any]	Optional filters to apply.	None
project_id	Optional[str]	The project that holds the run. Omitting it reads the project API key's project, which is deprecated.	None

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

def get_result(
    self,
    run_id: str,
    aggregate_function: Optional[str] = None,
    filters: Optional[Any] = None,
    *,
    project_id: Optional[str] = None,
) -> Dict[str, Any]:
    """Get experiment run result.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        run_id: The experiment run ID.
        aggregate_function: Aggregation function to apply. With
            ``project_id`` it must be one of ``average``, ``min``, ``max``,
            ``median``, ``p90``, ``p95``, ``p99``, ``sum`` or ``count``;
            any other value answers 400. Without ``project_id`` an
            unknown value falls back to ``average``.
        filters: Optional filters to apply.
        project_id: The project that holds the run. Omitting it reads the
            project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "get_result")
    return self._get_result(
        run_id, aggregate_function, filters, project_id=project_id
    )

compare_runs

compare_runs(
    new_run_id: str,
    old_run_id: str,
    aggregate_function: Optional[str] = None,
    filters: Optional[Any] = None,
    *,
    project_id: Optional[str] = None
) -> Dict[str, Any]

Compare two experiment runs.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
new_run_id	str	The new run ID to compare.	required
old_run_id	str	The old run ID to compare against.	required
aggregate_function	Optional[str]	Aggregation function to apply.	None
filters	Optional[Any]	Optional filters to apply.	None
project_id	Optional[str]	The project that holds both runs. Omitting it compares runs in the project API key's project, which is deprecated.	None

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

def compare_runs(
    self,
    new_run_id: str,
    old_run_id: str,
    aggregate_function: Optional[str] = None,
    filters: Optional[Any] = None,
    *,
    project_id: Optional[str] = None,
) -> Dict[str, Any]:
    """Compare two experiment runs.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        new_run_id: The new run ID to compare.
        old_run_id: The old run ID to compare against.
        aggregate_function: Aggregation function to apply.
        filters: Optional filters to apply.
        project_id: The project that holds both runs. Omitting it compares
            runs in the project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "compare_runs")
    return self._compare_runs(
        new_run_id,
        old_run_id,
        aggregate_function,
        filters,
        project_id=project_id,
    )

get_result_async async

get_result_async(
    run_id: str,
    aggregate_function: Optional[str] = None,
    filters: Optional[Any] = None,
    *,
    project_id: Optional[str] = None
) -> Dict[str, Any]

Get experiment run result asynchronously. See get_result().

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

async def get_result_async(
    self,
    run_id: str,
    aggregate_function: Optional[str] = None,
    filters: Optional[Any] = None,
    *,
    project_id: Optional[str] = None,
) -> Dict[str, Any]:
    """Get experiment run result asynchronously. See ``get_result()``.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "get_result_async")
    return await self._get_result_async(
        run_id, aggregate_function, filters, project_id=project_id
    )

compare_runs_async async

compare_runs_async(
    new_run_id: str,
    old_run_id: str,
    aggregate_function: Optional[str] = None,
    filters: Optional[Any] = None,
    *,
    project_id: Optional[str] = None
) -> Dict[str, Any]

Compare two experiment runs asynchronously.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
new_run_id	str	The new run ID to compare.	required
old_run_id	str	The old run ID to compare against.	required
aggregate_function	Optional[str]	Aggregation function to apply.	None
filters	Optional[Any]	Optional filters to apply.	None
project_id	Optional[str]	The project that holds both runs. Omitting it compares runs in the project API key's project, which is deprecated.	None

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

async def compare_runs_async(
    self,
    new_run_id: str,
    old_run_id: str,
    aggregate_function: Optional[str] = None,
    filters: Optional[Any] = None,
    *,
    project_id: Optional[str] = None,
) -> Dict[str, Any]:
    """Compare two experiment runs asynchronously.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        new_run_id: The new run ID to compare.
        old_run_id: The old run ID to compare against.
        aggregate_function: Aggregation function to apply.
        filters: Optional filters to apply.
        project_id: The project that holds both runs. Omitting it compares
            runs in the project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "compare_runs_async")
    return await self._compare_runs_async(
        new_run_id,
        old_run_id,
        aggregate_function,
        filters,
        project_id=project_id,
    )

get_run_result

get_run_result(
    run_id: str,
    aggregate_function: Optional[str] = None,
    filters: Optional[Any] = None,
    *,
    project_id: Optional[str] = None
) -> Dict[str, Any]

Get experiment run result (alias for get_result).

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

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

def get_run_result(
    self,
    run_id: str,
    aggregate_function: Optional[str] = None,
    filters: Optional[Any] = None,
    *,
    project_id: Optional[str] = None,
) -> Dict[str, Any]:
    """Get experiment run result (alias for get_result).

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "get_run_result")
    return self._get_result(
        run_id, aggregate_function, filters, project_id=project_id
    )

compare_run_events

compare_run_events(
    new_run_id: str,
    old_run_id: str,
    event_name: Optional[str] = None,
    event_type: Optional[str] = None,
    filter: Optional[Any] = None,
    limit: Optional[int] = None,
    page: Optional[int] = None,
    *,
    project_id: Optional[str] = None
) -> GetExperimentCompareEventsResponse

Compare events between two experiment runs.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
new_run_id	str	The new run ID to compare.	required
old_run_id	str	The old run ID to compare against.	required
event_name	Optional[str]	Filter by event name.	None
event_type	Optional[str]	Filter by event type.	None
filter	Optional[Any]	Additional filter criteria.	None
limit	Optional[int]	Maximum number of results.	None
page	Optional[int]	Page number for pagination.	None
project_id	Optional[str]	The project that holds both runs. Omitting it reads the project API key's project, which is deprecated.	None

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

def compare_run_events(
    self,
    new_run_id: str,
    old_run_id: str,
    event_name: Optional[str] = None,
    event_type: Optional[str] = None,
    filter: Optional[Any] = None,
    limit: Optional[int] = None,
    page: Optional[int] = None,
    *,
    project_id: Optional[str] = None,
) -> GetExperimentCompareEventsResponse:
    """Compare events between two experiment runs.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        new_run_id: The new run ID to compare.
        old_run_id: The old run ID to compare against.
        event_name: Filter by event name.
        event_type: Filter by event type.
        filter: Additional filter criteria.
        limit: Maximum number of results.
        page: Page number for pagination.
        project_id: The project that holds both runs. Omitting it reads the
            project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "compare_run_events")
    return self._compare_run_events(
        new_run_id,
        old_run_id,
        event_name,
        event_type,
        filter,
        limit,
        page,
        project_id=project_id,
    )

compare_run_events_async async

compare_run_events_async(
    new_run_id: str,
    old_run_id: str,
    event_name: Optional[str] = None,
    event_type: Optional[str] = None,
    filter: Optional[Any] = None,
    limit: Optional[int] = None,
    page: Optional[int] = None,
    *,
    project_id: Optional[str] = None
) -> GetExperimentCompareEventsResponse

Compare events between two experiment runs asynchronously.

Accepted keys

Data plane fine-grained API key with project_id; Project API key without project_id (deprecated)

Parameters:

Name	Type	Description	Default
new_run_id	str	The new run ID to compare.	required
old_run_id	str	The old run ID to compare against.	required
event_name	Optional[str]	Filter by event name.	None
event_type	Optional[str]	Filter by event type.	None
filter	Optional[Any]	Additional filter criteria.	None
limit	Optional[int]	Maximum number of results.	None
page	Optional[int]	Page number for pagination.	None
project_id	Optional[str]	The project that holds both runs. Omitting it reads the project API key's project, which is deprecated.	None

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

async def compare_run_events_async(
    self,
    new_run_id: str,
    old_run_id: str,
    event_name: Optional[str] = None,
    event_type: Optional[str] = None,
    filter: Optional[Any] = None,
    limit: Optional[int] = None,
    page: Optional[int] = None,
    *,
    project_id: Optional[str] = None,
) -> GetExperimentCompareEventsResponse:
    """Compare events between two experiment runs asynchronously.

    Accepted keys:
        Data plane fine-grained API key with ``project_id``; Project API key
        without ``project_id`` (deprecated)

    Args:
        new_run_id: The new run ID to compare.
        old_run_id: The old run ID to compare against.
        event_name: Filter by event name.
        event_type: Filter by event type.
        filter: Additional filter criteria.
        limit: Maximum number of results.
        page: Page number for pagination.
        project_id: The project that holds both runs. Omitting it reads the
            project API key's project, which is deprecated.
    """
    if project_id is None:
        _warn_call_without_project_id("experiments", "compare_run_events_async")
    return await self._compare_run_events_async(
        new_run_id,
        old_run_id,
        event_name,
        event_type,
        filter,
        limit,
        page,
        project_id=project_id,
    )

MetricsAPI

Bases: BaseAPI

Metrics API.

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

class MetricsAPI(BaseAPI):
    """Metrics API."""

    # Sync methods
    def list(
        self,
        project: Optional[str] = None,
        name: Optional[str] = None,
        type: Optional[str] = None,
    ) -> List[MetricItem]:
        """List metrics.

        Accepted keys:
            Project API key
        """
        # Keep the legacy filters in the public signature for back-compat, but
        # warn because the dataplane metrics list endpoint no longer honors them.
        if project is not None or name is not None:
            warnings.warn(
                "The 'project' and 'name' parameters are no longer supported for "
                "metrics.list() and will be removed in v2.0.",
                DeprecationWarning,
                stacklevel=2,
            )
        # The regenerated client only supports filtering by metric type or ID.
        del project, name
        # Preserve the high-level SDK list surface by unwrapping the transport envelope.
        response = metrics_svc.getMetrics(self._api_config, type=type)
        return response.metrics

    def create(self, request: CreateMetricRequest) -> CreateMetricResponse:
        """Create a metric.

        Accepted keys:
            Project API key
        """
        return metrics_svc.createMetric(self._api_config, data=request)

    def update(self, request: UpdateMetricRequest) -> UpdateMetricResponse:
        """Update a metric.

        Accepted keys:
            Project API key
        """
        return metrics_svc.updateMetricLegacy(self._api_config, data=request)

    def delete(self, id: str) -> DeleteMetricResponse:
        """Delete a metric.

        Accepted keys:
            Project API key
        """
        return metrics_svc.deleteMetricLegacy(self._api_config, metric_id=id)

    # Async methods
    async def list_async(
        self,
        project: Optional[str] = None,
        name: Optional[str] = None,
        type: Optional[str] = None,
    ) -> List[MetricItem]:
        """List metrics asynchronously.

        Accepted keys:
            Project API key
        """
        # Keep the legacy filters in the public signature for back-compat, but
        # warn because the dataplane metrics list endpoint no longer honors them.
        if project is not None or name is not None:
            warnings.warn(
                "The 'project' and 'name' parameters are no longer supported for "
                "metrics.list() and will be removed in v2.0.",
                DeprecationWarning,
                stacklevel=2,
            )
        del project, name
        response = await metrics_svc_async.getMetrics(self._api_config, type=type)
        return response.metrics

    async def create_async(self, request: CreateMetricRequest) -> CreateMetricResponse:
        """Create a metric asynchronously.

        Accepted keys:
            Project API key
        """
        return await metrics_svc_async.createMetric(self._api_config, data=request)

    async def update_async(self, request: UpdateMetricRequest) -> UpdateMetricResponse:
        """Update a metric asynchronously.

        Accepted keys:
            Project API key
        """
        return await metrics_svc_async.updateMetricLegacy(
            self._api_config, data=request
        )

    async def delete_async(self, id: str) -> DeleteMetricResponse:
        """Delete a metric asynchronously.

        Accepted keys:
            Project API key
        """
        return await metrics_svc_async.deleteMetricLegacy(
            self._api_config, metric_id=id
        )

    # Backwards compatible aliases
    def get_metric(self, id: str) -> List[MetricItem]:
        """Get a metric (backwards compatible alias).

        Accepted keys:
            Project API key
        """
        response = metrics_svc.getMetrics(self._api_config, id=id)
        return response.metrics

    def create_metric(self, request: CreateMetricRequest) -> CreateMetricResponse:
        """Create a metric (backwards compatible alias).

        Accepted keys:
            Project API key
        """
        return self.create(request)

    def update_metric(self, request: UpdateMetricRequest) -> UpdateMetricResponse:
        """Update a metric (backwards compatible alias).

        Accepted keys:
            Project API key
        """
        return self.update(request)

    def delete_metric(self, id: str) -> DeleteMetricResponse:
        """Delete a metric (backwards compatible alias).

        Accepted keys:
            Project API key
        """
        return self.delete(id)

    def list_metrics(
        self,
        project: Optional[str] = None,
        name: Optional[str] = None,
        type: Optional[str] = None,
    ) -> List[MetricItem]:
        """List metrics (backwards compatible alias).

        Accepted keys:
            Project API key
        """
        return self.list(project=project, name=name, type=type)

list

list(
    project: Optional[str] = None,
    name: Optional[str] = None,
    type: Optional[str] = None,
) -> List[MetricItem]

List metrics.

Accepted keys

Project API key

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

def list(
    self,
    project: Optional[str] = None,
    name: Optional[str] = None,
    type: Optional[str] = None,
) -> List[MetricItem]:
    """List metrics.

    Accepted keys:
        Project API key
    """
    # Keep the legacy filters in the public signature for back-compat, but
    # warn because the dataplane metrics list endpoint no longer honors them.
    if project is not None or name is not None:
        warnings.warn(
            "The 'project' and 'name' parameters are no longer supported for "
            "metrics.list() and will be removed in v2.0.",
            DeprecationWarning,
            stacklevel=2,
        )
    # The regenerated client only supports filtering by metric type or ID.
    del project, name
    # Preserve the high-level SDK list surface by unwrapping the transport envelope.
    response = metrics_svc.getMetrics(self._api_config, type=type)
    return response.metrics

create

create(
    request: CreateMetricRequest,
) -> CreateMetricResponse

Create a metric.

Accepted keys

Project API key

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

def create(self, request: CreateMetricRequest) -> CreateMetricResponse:
    """Create a metric.

    Accepted keys:
        Project API key
    """
    return metrics_svc.createMetric(self._api_config, data=request)

update

update(
    request: UpdateMetricRequest,
) -> UpdateMetricResponse

Update a metric.

Accepted keys

Project API key

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

def update(self, request: UpdateMetricRequest) -> UpdateMetricResponse:
    """Update a metric.

    Accepted keys:
        Project API key
    """
    return metrics_svc.updateMetricLegacy(self._api_config, data=request)

delete

delete(id: str) -> DeleteMetricResponse

Delete a metric.

Accepted keys

Project API key

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

def delete(self, id: str) -> DeleteMetricResponse:
    """Delete a metric.

    Accepted keys:
        Project API key
    """
    return metrics_svc.deleteMetricLegacy(self._api_config, metric_id=id)

list_async async

list_async(
    project: Optional[str] = None,
    name: Optional[str] = None,
    type: Optional[str] = None,
) -> List[MetricItem]

List metrics asynchronously.

Accepted keys

Project API key

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

async def list_async(
    self,
    project: Optional[str] = None,
    name: Optional[str] = None,
    type: Optional[str] = None,
) -> List[MetricItem]:
    """List metrics asynchronously.

    Accepted keys:
        Project API key
    """
    # Keep the legacy filters in the public signature for back-compat, but
    # warn because the dataplane metrics list endpoint no longer honors them.
    if project is not None or name is not None:
        warnings.warn(
            "The 'project' and 'name' parameters are no longer supported for "
            "metrics.list() and will be removed in v2.0.",
            DeprecationWarning,
            stacklevel=2,
        )
    del project, name
    response = await metrics_svc_async.getMetrics(self._api_config, type=type)
    return response.metrics

create_async async

create_async(
    request: CreateMetricRequest,
) -> CreateMetricResponse

Create a metric asynchronously.

Accepted keys

Project API key

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

async def create_async(self, request: CreateMetricRequest) -> CreateMetricResponse:
    """Create a metric asynchronously.

    Accepted keys:
        Project API key
    """
    return await metrics_svc_async.createMetric(self._api_config, data=request)

update_async async

update_async(
    request: UpdateMetricRequest,
) -> UpdateMetricResponse

Update a metric asynchronously.

Accepted keys

Project API key

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

async def update_async(self, request: UpdateMetricRequest) -> UpdateMetricResponse:
    """Update a metric asynchronously.

    Accepted keys:
        Project API key
    """
    return await metrics_svc_async.updateMetricLegacy(
        self._api_config, data=request
    )

delete_async async

delete_async(id: str) -> DeleteMetricResponse

Delete a metric asynchronously.

Accepted keys

Project API key

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

async def delete_async(self, id: str) -> DeleteMetricResponse:
    """Delete a metric asynchronously.

    Accepted keys:
        Project API key
    """
    return await metrics_svc_async.deleteMetricLegacy(
        self._api_config, metric_id=id
    )

get_metric

get_metric(id: str) -> List[MetricItem]

Get a metric (backwards compatible alias).

Accepted keys

Project API key

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

def get_metric(self, id: str) -> List[MetricItem]:
    """Get a metric (backwards compatible alias).

    Accepted keys:
        Project API key
    """
    response = metrics_svc.getMetrics(self._api_config, id=id)
    return response.metrics

create_metric

create_metric(
    request: CreateMetricRequest,
) -> CreateMetricResponse

Create a metric (backwards compatible alias).

Accepted keys

Project API key

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

def create_metric(self, request: CreateMetricRequest) -> CreateMetricResponse:
    """Create a metric (backwards compatible alias).

    Accepted keys:
        Project API key
    """
    return self.create(request)

update_metric

update_metric(
    request: UpdateMetricRequest,
) -> UpdateMetricResponse

Update a metric (backwards compatible alias).

Accepted keys

Project API key

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

def update_metric(self, request: UpdateMetricRequest) -> UpdateMetricResponse:
    """Update a metric (backwards compatible alias).

    Accepted keys:
        Project API key
    """
    return self.update(request)

delete_metric

delete_metric(id: str) -> DeleteMetricResponse

Delete a metric (backwards compatible alias).

Accepted keys

Project API key

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

def delete_metric(self, id: str) -> DeleteMetricResponse:
    """Delete a metric (backwards compatible alias).

    Accepted keys:
        Project API key
    """
    return self.delete(id)

list_metrics

list_metrics(
    project: Optional[str] = None,
    name: Optional[str] = None,
    type: Optional[str] = None,
) -> List[MetricItem]

List metrics (backwards compatible alias).

Accepted keys

Project API key

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

def list_metrics(
    self,
    project: Optional[str] = None,
    name: Optional[str] = None,
    type: Optional[str] = None,
) -> List[MetricItem]:
    """List metrics (backwards compatible alias).

    Accepted keys:
        Project API key
    """
    return self.list(project=project, name=name, type=type)

MetricVersionsAPI

Bases: BaseAPI

Metric Versions API.

Each metric has an immutable history of versions; one version at a time is marked as deployed (used by evaluations). Versions are nested under the parent metric at /v1/metrics/{metric_id}/versions.

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

class MetricVersionsAPI(BaseAPI):
    """Metric Versions API.

    Each metric has an immutable history of versions; one version at a time is
    marked as deployed (used by evaluations). Versions are nested under the
    parent metric at ``/v1/metrics/{metric_id}/versions``.
    """

    # Sync methods
    def list(self, metric_id: str) -> GetMetricVersionsResponse:
        """List all versions for a metric, oldest-first.

        Accepted keys:
            Project API key
        """
        return metric_versions_svc.getMetricVersions(
            self._api_config, metric_id=metric_id
        )

    def create(
        self, metric_id: str, request: CreateMetricVersionRequest
    ) -> CreateMetricVersionResponse:
        """Create a new version of a metric.

        Set ``request.deploy_immediately = True`` to atomically mark the new
        version as deployed in the same transaction.

        Accepted keys:
            Project API key
        """
        return metric_versions_svc.createMetricVersion(
            self._api_config, metric_id=metric_id, data=request
        )

    def deploy(self, metric_id: str, version_name: str) -> DeployMetricVersionResponse:
        """Deploy an existing version of a metric, replacing the live version.

        Accepted keys:
            Project API key
        """
        return metric_versions_svc.deployMetricVersion(
            self._api_config, metric_id=metric_id, version_name=version_name
        )

    # Async methods
    async def list_async(self, metric_id: str) -> GetMetricVersionsResponse:
        """List all versions for a metric asynchronously.

        Accepted keys:
            Project API key
        """
        return await metric_versions_svc_async.getMetricVersions(
            self._api_config, metric_id=metric_id
        )

    async def create_async(
        self, metric_id: str, request: CreateMetricVersionRequest
    ) -> CreateMetricVersionResponse:
        """Create a new version of a metric asynchronously.

        Accepted keys:
            Project API key
        """
        return await metric_versions_svc_async.createMetricVersion(
            self._api_config, metric_id=metric_id, data=request
        )

    async def deploy_async(
        self, metric_id: str, version_name: str
    ) -> DeployMetricVersionResponse:
        """Deploy an existing version of a metric asynchronously.

        Accepted keys:
            Project API key
        """
        return await metric_versions_svc_async.deployMetricVersion(
            self._api_config, metric_id=metric_id, version_name=version_name
        )

list

list(metric_id: str) -> GetMetricVersionsResponse

List all versions for a metric, oldest-first.

Accepted keys

Project API key

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

def list(self, metric_id: str) -> GetMetricVersionsResponse:
    """List all versions for a metric, oldest-first.

    Accepted keys:
        Project API key
    """
    return metric_versions_svc.getMetricVersions(
        self._api_config, metric_id=metric_id
    )

create

create(
    metric_id: str, request: CreateMetricVersionRequest
) -> CreateMetricVersionResponse

Create a new version of a metric.

Set request.deploy_immediately = True to atomically mark the new version as deployed in the same transaction.

Accepted keys

Project API key

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

def create(
    self, metric_id: str, request: CreateMetricVersionRequest
) -> CreateMetricVersionResponse:
    """Create a new version of a metric.

    Set ``request.deploy_immediately = True`` to atomically mark the new
    version as deployed in the same transaction.

    Accepted keys:
        Project API key
    """
    return metric_versions_svc.createMetricVersion(
        self._api_config, metric_id=metric_id, data=request
    )

deploy

deploy(
    metric_id: str, version_name: str
) -> DeployMetricVersionResponse

Deploy an existing version of a metric, replacing the live version.

Accepted keys

Project API key

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

def deploy(self, metric_id: str, version_name: str) -> DeployMetricVersionResponse:
    """Deploy an existing version of a metric, replacing the live version.

    Accepted keys:
        Project API key
    """
    return metric_versions_svc.deployMetricVersion(
        self._api_config, metric_id=metric_id, version_name=version_name
    )

list_async async

list_async(metric_id: str) -> GetMetricVersionsResponse

List all versions for a metric asynchronously.

Accepted keys

Project API key

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

async def list_async(self, metric_id: str) -> GetMetricVersionsResponse:
    """List all versions for a metric asynchronously.

    Accepted keys:
        Project API key
    """
    return await metric_versions_svc_async.getMetricVersions(
        self._api_config, metric_id=metric_id
    )

create_async async

create_async(
    metric_id: str, request: CreateMetricVersionRequest
) -> CreateMetricVersionResponse

Create a new version of a metric asynchronously.

Accepted keys

Project API key

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

async def create_async(
    self, metric_id: str, request: CreateMetricVersionRequest
) -> CreateMetricVersionResponse:
    """Create a new version of a metric asynchronously.

    Accepted keys:
        Project API key
    """
    return await metric_versions_svc_async.createMetricVersion(
        self._api_config, metric_id=metric_id, data=request
    )

deploy_async async

deploy_async(
    metric_id: str, version_name: str
) -> DeployMetricVersionResponse

Deploy an existing version of a metric asynchronously.

Accepted keys

Project API key

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

async def deploy_async(
    self, metric_id: str, version_name: str
) -> DeployMetricVersionResponse:
    """Deploy an existing version of a metric asynchronously.

    Accepted keys:
        Project API key
    """
    return await metric_versions_svc_async.deployMetricVersion(
        self._api_config, metric_id=metric_id, version_name=version_name
    )

SessionsAPI

Bases: BaseAPI

Sessions API.

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

class SessionsAPI(BaseAPI):
    """Sessions API."""

    # The canonical spec only exposes session creation here. Session reads should
    # go through the event query/export helpers, keyed by session_id.

    @staticmethod
    def _coerce_start_session_request(
        data: Union[StartSessionRequest, Dict[str, Any]],
    ) -> StartSessionRequest:
        """Normalize session start inputs to the generated request model."""
        if isinstance(data, StartSessionRequest):
            return data

        if "session" in data:
            return StartSessionRequest(**data)

        # Older SDK callers pass the raw session payload directly. Keep accepting
        # that flat shape and wrap it into the nested request model for back-compat.
        warnings.warn(
            "Passing a flat session payload to sessions.start() is deprecated and "
            "will be removed in v2.0. Pass StartSessionRequest(session=...) or "
            "{'session': {...}} instead.",
            DeprecationWarning,
            stacklevel=3,
        )
        return StartSessionRequest.model_validate({"session": data})

    def start(
        self, data: Union[StartSessionRequest, Dict[str, Any]]
    ) -> PostSessionStartResponse:
        """Start a new session.

        Accepted keys:
            Ingestion API key; Project API key (deprecated)
        """
        request = self._coerce_start_session_request(data)
        return sessions_svc.startSessionLegacy(self._ingestion_api_config, data=request)

    # Async methods
    async def start_async(
        self, data: Union[StartSessionRequest, Dict[str, Any]]
    ) -> PostSessionStartResponse:
        """Start a new session asynchronously.

        Accepted keys:
            Ingestion API key; Project API key (deprecated)
        """
        request = self._coerce_start_session_request(data)
        return await sessions_svc_async.startSessionLegacy(
            self._ingestion_api_config, data=request
        )

    # Backwards compatible aliases
    def create_session(
        self, request: Union[StartSessionRequest, Dict[str, Any]]
    ) -> PostSessionStartResponse:
        """Create/start a session (backwards compatible alias for start()).

        Accepted keys:
            Ingestion API key; Project API key (deprecated)
        """
        return self.start(request)

    def start_session(
        self, request: Union[StartSessionRequest, Dict[str, Any]]
    ) -> PostSessionStartResponse:
        """Start a session (backwards compatible alias for start()).

        Accepted keys:
            Ingestion API key; Project API key (deprecated)
        """
        return self.start(request)

start

start(
    data: Union[StartSessionRequest, Dict[str, Any]],
) -> PostSessionStartResponse

Start a new session.

Accepted keys

Ingestion API key; Project API key (deprecated)

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

def start(
    self, data: Union[StartSessionRequest, Dict[str, Any]]
) -> PostSessionStartResponse:
    """Start a new session.

    Accepted keys:
        Ingestion API key; Project API key (deprecated)
    """
    request = self._coerce_start_session_request(data)
    return sessions_svc.startSessionLegacy(self._ingestion_api_config, data=request)

start_async async

start_async(
    data: Union[StartSessionRequest, Dict[str, Any]],
) -> PostSessionStartResponse

Start a new session asynchronously.

Accepted keys

Ingestion API key; Project API key (deprecated)

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

async def start_async(
    self, data: Union[StartSessionRequest, Dict[str, Any]]
) -> PostSessionStartResponse:
    """Start a new session asynchronously.

    Accepted keys:
        Ingestion API key; Project API key (deprecated)
    """
    request = self._coerce_start_session_request(data)
    return await sessions_svc_async.startSessionLegacy(
        self._ingestion_api_config, data=request
    )

create_session

create_session(
    request: Union[StartSessionRequest, Dict[str, Any]],
) -> PostSessionStartResponse

Create/start a session (backwards compatible alias for start()).

Accepted keys

Ingestion API key; Project API key (deprecated)

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

def create_session(
    self, request: Union[StartSessionRequest, Dict[str, Any]]
) -> PostSessionStartResponse:
    """Create/start a session (backwards compatible alias for start()).

    Accepted keys:
        Ingestion API key; Project API key (deprecated)
    """
    return self.start(request)

start_session

start_session(
    request: Union[StartSessionRequest, Dict[str, Any]],
) -> PostSessionStartResponse

Start a session (backwards compatible alias for start()).

Accepted keys

Ingestion API key; Project API key (deprecated)

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

def start_session(
    self, request: Union[StartSessionRequest, Dict[str, Any]]
) -> PostSessionStartResponse:
    """Start a session (backwards compatible alias for start()).

    Accepted keys:
        Ingestion API key; Project API key (deprecated)
    """
    return self.start(request)

HoneyHive

Main HoneyHive API client.

Provides an ergonomic interface to the HoneyHive API with both sync and async methods.

Each method's "Accepted keys" section names the API keys it can send. A project API key is passed as api_key, an ingestion API key as ingestion_api_key, and a data plane fine-grained API key as data_plane_api_key. Where a method names a typed key beside a project API key, the project API key is deprecated. Each key type, and where to create one, is described at https://docs.honeyhive.ai/v2/workspace/api-keys#key-types.

Example::

client = HoneyHive(api_key="your-api-key")

# Sync
configs = client.configurations.list()

# Async
configs = await client.configurations.list_async()

Attributes:

Name	Type	Description
configurations		API for managing configurations.
datapoints		API for managing datapoints.
datasets		API for managing datasets.
events		API for managing events.
experiments		API for managing experiment runs.
metrics		API for managing metrics.
sessions		API for managing sessions.

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

class HoneyHive:
    """Main HoneyHive API client.

    Provides an ergonomic interface to the HoneyHive API with both
    sync and async methods.

    Each method's "Accepted keys" section names the API keys it can send. A
    project API key is passed as ``api_key``, an ingestion API key as
    ``ingestion_api_key``, and a data plane fine-grained API key as
    ``data_plane_api_key``. Where a method names a typed key beside a project
    API key, the project API key is deprecated. Each key type, and where to
    create one, is described at
    <https://docs.honeyhive.ai/v2/workspace/api-keys#key-types>.

    Example::

        client = HoneyHive(api_key="your-api-key")

        # Sync
        configs = client.configurations.list()

        # Async
        configs = await client.configurations.list_async()

    Attributes:
        configurations: API for managing configurations.
        datapoints: API for managing datapoints.
        datasets: API for managing datasets.
        events: API for managing events.
        experiments: API for managing experiment runs.
        metrics: API for managing metrics.
        sessions: API for managing sessions.
    """

    # The published constructor keeps every parameter it has ever accepted until
    # v2.0, so its argument count is fixed by compatibility and the suppression
    # below stays until then. This method therefore holds only the frozen
    # signature and the deprecation warnings; the construction logic lives in
    # _initialize, where the argument limit and every other lint still apply.
    # pylint: disable-next=too-many-arguments
    def __init__(
        self,
        api_key: Optional[str] = None,
        # `project` sits at positional slot #2 to mirror the legacy SDK shape
        # — pre-1.0 callers wrote `HoneyHive("key", "project-name")` positionally.
        # Keeping the slot reserved means those calls still execute (with a
        # DeprecationWarning) instead of binding the value to `base_url` or
        # raising TypeError. The argument is otherwise ignored.
        project: Optional[str] = None,
        *,
        ingestion_api_key: Optional[str] = None,
        data_plane_api_key: Optional[str] = None,
        # Primary URL parameter
        base_url: Optional[str] = None,
        # Backwards compatible alias for base_url
        server_url: Optional[str] = None,
        cp_base_url: Optional[str] = None,
        timeout: Optional[float] = None,
        retry_config: Optional[Any] = None,
        rate_limit_calls: Optional[int] = None,
        rate_limit_window: Optional[float] = None,
        max_connections: Optional[int] = None,
        max_keepalive: Optional[int] = None,
        test_mode: Optional[bool] = None,
        verbose: Optional[bool] = None,
        tracer_instance: Optional[Any] = None,
    ) -> None:
        """Initialize the HoneyHive client.

        Deprecated parameters are accepted so existing calls keep working; each
        one passed emits a ``DeprecationWarning`` and is otherwise ignored. The
        client is built by ``_initialize`` from the parameters still in use.

        Args:
            api_key: HoneyHive API key (typically starts with ``hh_``).
                     Falls back to HH_API_KEY environment variable.
            ingestion_api_key: Ingestion API key (starts with ``hh_ingst_``) sent
                     with requests that create sessions or write events. Falls
                     back to the HH_INGESTION_API_KEY environment variable, then
                     to api_key. A value that is set but is not an ingestion key
                     raises ValueError here, at construction.
            data_plane_api_key: Fine-grained data plane API key (starts with
                     ``hh_fgdp_``), the credential for the operations that name
                     a project in their path. Falls back to the
                     HH_DATA_PLANE_API_KEY environment variable. Without either,
                     those operations raise MissingApiKeyError before sending,
                     because they refuse api_key. A value that is set but is not
                     a fine-grained data plane key raises ValueError here, at
                     construction.
            project: Deprecated and ignored; will be removed in v2.0. The
                backend infers project context from the API key and session.
            base_url: API base URL for HoneyHive.
                      Falls back to HH_API_URL env var, then https://api.dp1.us.honeyhive.ai.
            server_url: Alias for base_url, kept for callers written against the
                older name; base_url wins when both are given.
            cp_base_url: Deprecated and ignored; will be removed in v2.0. The
                SDK uses a single base_url for all operations.
            timeout: Request timeout in seconds. Falls back to the HH_API_TIMEOUT
                env var, then to the SDK default of 5s. Pass a larger value when
                fetching large payloads (e.g. datasets.list for many datasets).
            retry_config: Retry configuration for ``events.export()`` /
                ``events.export_async()``, the only methods that retry. A
                ``RetryConfig`` is used as-is. When omitted, ``HH_MAX_RETRIES``
                sets the retry budget (default 3).
            rate_limit_calls: Deprecated and ignored; will be removed in v2.0.
            rate_limit_window: Deprecated and ignored; will be removed in v2.0.
            max_connections: Deprecated and ignored; will be removed in v2.0.
            max_keepalive: Deprecated and ignored; will be removed in v2.0.
            test_mode: Recorded and exposed as ``HoneyHive.test_mode``.
            verbose: Recorded and exposed as ``HoneyHive.verbose``.
            tracer_instance: Deprecated and ignored; will be removed in v2.0.
        """
        deprecated_parameters = (
            ("project", project),
            ("cp_base_url", cp_base_url),
            ("rate_limit_calls", rate_limit_calls),
            ("rate_limit_window", rate_limit_window),
            ("max_connections", max_connections),
            ("max_keepalive", max_keepalive),
            ("tracer_instance", tracer_instance),
        )
        for name, value in deprecated_parameters:
            if value is not None:
                message = (
                    f"The {name!r} parameter to HoneyHive() is deprecated and "
                    "ignored; it will be removed in v2.0. Remove it from "
                    "HoneyHive() calls."
                )
                warnings.warn(message, DeprecationWarning, stacklevel=2)
                logger.warning(message)

        self._initialize(
            api_key=api_key,
            ingestion_api_key=ingestion_api_key,
            data_plane_api_key=data_plane_api_key,
            base_url=base_url,
            server_url=server_url,
            timeout=timeout,
            retry_config=retry_config,
            test_mode=test_mode,
            verbose=verbose,
        )

    def _initialize(
        self,
        *,
        api_key: Optional[str],
        ingestion_api_key: Optional[str],
        data_plane_api_key: Optional[str],
        base_url: Optional[str],
        server_url: Optional[str],
        timeout: Optional[float],
        retry_config: Optional[Any],
        test_mode: Optional[bool],
        verbose: Optional[bool],
    ) -> None:
        """Build the client from the parameters still in use.

        ``__init__`` holds the published signature and drops the deprecated
        parameters before calling here.
        """
        connection = ResolvedConfig.resolve(
            api_key=api_key,
            ingestion_api_key=ingestion_api_key,
            data_plane_api_key=data_plane_api_key,
            api_url=base_url or server_url,
        )
        self._api_key = connection.api_key or ""
        self._ingestion_api_key = connection.ingestion_api_key or ""
        self._data_plane_api_key = connection.data_plane_api_key or ""
        resolved_base_url = connection.api_url

        # Exposed through the timeout, test_mode, verbose and retry_config properties.
        self._timeout = timeout
        self._test_mode = test_mode if test_mode is not None else False
        self._verbose = verbose if verbose is not None else False
        self._retry_config = RetryConfig.from_optional(retry_config)

        # Create API config. The request timeout is resolved from the explicit
        # arg > HH_API_TIMEOUT env var; when neither is set we omit the key so
        # the APIConfig default (5.0s) is preserved. Note: passing timeout=None
        # to HoneyHive() keeps the default (None means "unset" here), whereas
        # APIConfig(timeout=None) disables timeouts at the low level.
        api_config_kwargs: Dict[str, Any] = {
            "base_path": resolved_base_url,
            "access_token": self._api_key,
        }
        resolved_timeout = _resolve_api_timeout(timeout)
        if resolved_timeout is not None:
            api_config_kwargs["timeout"] = resolved_timeout
        self._api_config = APIConfig(**api_config_kwargs)

        # Only the namespaces with wrappers for a typed key's operations take
        # that key; they derive the configuration for those requests from
        # `_api_config` and the key at call time (see
        # BaseAPI._typed_api_config), so there is one configuration object for
        # a caller to mutate. Every other namespace sends `_api_config` as it
        # always has.
        self.charts = ChartsAPI(
            self._api_config, data_plane_api_key=self._data_plane_api_key
        )
        self.configurations = ConfigurationsAPI(self._api_config)
        self.datapoints = DatapointsAPI(
            self._api_config, data_plane_api_key=self._data_plane_api_key
        )
        self.datasets = DatasetsAPI(
            self._api_config, data_plane_api_key=self._data_plane_api_key
        )
        self.events = EventsAPI(
            self._api_config,
            self._ingestion_api_key,
            retry_config=self._retry_config,
        )
        self.experiments = ExperimentsAPI(
            self._api_config, data_plane_api_key=self._data_plane_api_key
        )
        self.metrics = MetricsAPI(self._api_config)
        self.metric_versions = MetricVersionsAPI(self._api_config)
        self.sessions = SessionsAPI(self._api_config, self._ingestion_api_key)

        # Alias for backwards compatibility
        self.evaluations = self.experiments

    @property
    def test_mode(self) -> bool:
        """Return whether client is in test mode."""
        return self._test_mode

    @property
    def verbose(self) -> bool:
        """Return whether verbose mode is enabled."""
        return self._verbose

    @property
    def timeout(self) -> Optional[float]:
        """Return the configured timeout."""
        return self._timeout

    @property
    def retry_config(self) -> RetryConfig:
        """Retry configuration used by ``events.export()`` / ``export_async()``."""
        return self._retry_config

    @property
    def api_config(self) -> APIConfig:
        """Access the underlying API configuration."""
        return self._api_config

    @property
    def api_key(self) -> str:
        """Get the HoneyHive API key."""
        return self._api_key

    @property
    def ingestion_api_key(self) -> str:
        """The ingestion API key, or ``""`` when api_key is used for ingestion too."""
        return self._ingestion_api_key

    @property
    def data_plane_api_key(self) -> str:
        """The fine-grained data plane API key, or ``""`` when none is configured."""
        return self._data_plane_api_key

    @property
    def server_url(self) -> str:
        """Get the HoneyHive API server URL."""
        return self._api_config.base_path

    @server_url.setter
    def server_url(self, value: str) -> None:
        """Set the HoneyHive API server URL."""
        self._api_config.base_path = value

test_mode property

test_mode: bool

Return whether client is in test mode.

verbose property

verbose: bool

Return whether verbose mode is enabled.

timeout property

timeout: Optional[float]

Return the configured timeout.

retry_config property

retry_config: RetryConfig

Retry configuration used by events.export() / export_async().

api_config property

api_config: APIConfig

Access the underlying API configuration.

api_key property

api_key: str

Get the HoneyHive API key.

ingestion_api_key property

ingestion_api_key: str

The ingestion API key, or "" when api_key is used for ingestion too.

data_plane_api_key property

data_plane_api_key: str

The fine-grained data plane API key, or "" when none is configured.

server_url property writable

server_url: str

Get the HoneyHive API server URL.