Skip to main content

​
synth_ai.sdk.managed_pools

Helpers for managed environment pools (data sources + uploads). These utilities generate presigned uploads and register pool data sources. They are plan-gated and will raise PlanGatingError when the account is not eligible. Example: 
>>> from synth_ai.sdk.managed_pools import (
...     create_managed_pool_upload_url,
...     upload_managed_pool_file,
...     create_managed_pool_upload_data_source,
... )
>>>
>>> upload = create_managed_pool_upload_url(filename="tasks.zip")
>>> upload_managed_pool_file(upload["upload_url"], "./tasks.zip")
>>> data_source = create_managed_pool_upload_data_source(
...     upload_id=upload["upload_id"],
...     upload_key=upload["upload_key"],
... )
>>> print(data_source["data_source_id"])

​
Functions

​
create_managed_pool_upload_url

create_managed_pool_upload_url(
    *,
    backend_base: str | None = None,
    api_key: str | None = None,
    filename: str | None = None,
    content_type: str | None = None,
    expires_in_seconds: int | None = None,
    timeout: float = 30.0,
) -> dict[str, Any]
Request a presigned upload URL for managed pool data.

​
upload_managed_pool_bytes

upload_managed_pool_bytes(
    upload_url: str,
    data: bytes,
    *,
    content_type: str | None = None,
    timeout: float = 60.0,
) -> None
Upload raw bytes to a managed pool presigned URL.

​
upload_managed_pool_file

upload_managed_pool_file(
    upload_url: str,
    file_path: str | Path,
    *,
    content_type: str | None = None,
    timeout: float = 60.0,
) -> None
Upload a local file to a managed pool presigned URL.

​
create_managed_pool_upload_data_source

create_managed_pool_upload_data_source(
    *,
    backend_base: str | None = None,
    api_key: str | None = None,
    upload_id: str,
    upload_key: str,
    timeout: float = 30.0,
) -> dict[str, Any]
Create a managed pool data source from a presigned upload.

​
create_managed_pool_s3_data_source

create_managed_pool_s3_data_source(
    *,
    backend_base: str | None = None,
    crypto_base: str | None = None,
    api_key: str | None = None,
    bucket: str,
    prefix: str,
    credentials: Mapping[str, str | None],
    timeout: float = 30.0,
) -> dict[str, Any]
Create a managed pool S3 data source (with encrypted credentials).