Skip to main content
Version: v2.0

aixplain.v2.resource

Resource management module for v2 API.

with_hooks​

def with_hooks(func: Callable) -> Callable

[view_source]

Generic decorator to add before/after hooks to resource operations.

This decorator automatically infers the operation name from the function name and provides a consistent pattern for all operations:

  • Before hooks can return early to bypass the operation
  • After hooks can transform the result
  • Error handling is consistent across all operations
  • Supports both positional and keyword arguments

Usage:

@with_hooks
def save(self, **kwargs):
    # operation implementation

@with_hooks
def run(self, *args, **kwargs):
    # operation implementation with positional args

encode_resource_id​

def encode_resource_id(resource_id: str) -> str

[view_source]

URL encode a resource ID for use in API paths.

Arguments:

  • resource_id - The resource ID to encode

Returns:

The URL-encoded resource ID

HasContext Objects​

@runtime_checkable
class HasContext(Protocol)

[view_source]

Protocol for classes that have a context attribute.

HasResourcePath Objects​

@runtime_checkable
class HasResourcePath(Protocol)

[view_source]

Protocol for classes that have a RESOURCE_PATH attribute.

HasFromDict Objects​

@runtime_checkable
class HasFromDict(Protocol)

[view_source]

Protocol for classes that have a from_dict method.

from_dict​

@classmethod
def from_dict(cls: type, data: dict) -> Any

[view_source]

Create an instance from a dictionary.

HasToDict Objects​

@runtime_checkable
class HasToDict(Protocol)

[view_source]

Protocol for classes that have a to_dict method.

to_dict​

def to_dict() -> dict

[view_source]

Convert instance to dictionary.

BaseMixin Objects​

class BaseMixin()

[view_source]

Base mixin with meta capabilities for resource operations.

__init_subclass__​

def __init_subclass__(cls: type, **kwargs: Any) -> None

[view_source]

Initialize subclass with validation.

BaseResource Objects​

@dataclass_json

@dataclass
class BaseResource()

[view_source]

Base class for all resources.

Attributes:

  • context - The Aixplain client instance (hidden from serialization).
  • RESOURCE_PATH - The API resource path.
  • id - The resource ID.
  • name - The resource name.
  • description - The resource description.
  • path - Full path identifier (e.g., "openai/whisper-large/groq").

path​

Full path e.g. "openai/whisper-large/groq"

is_modified​

@property
def is_modified() -> bool

[view_source]

Check if the resource has been modified since last save.

Returns:

  • bool - True if the resource has been modified, False otherwise

is_deleted​

@property
def is_deleted() -> bool

[view_source]

Check if the resource has been deleted.

Returns:

  • bool - True if the resource has been deleted, False otherwise

before_save​

def before_save(*args: Any, **kwargs: Any) -> Optional[dict]

[view_source]

Optional callback called before the resource is saved.

Override this method to add custom logic before saving.

Arguments:

  • *args - Positional arguments passed to the save operation
  • **kwargs - Keyword arguments passed to the save operation

Returns:

  • Optional[dict] - If not None, this result will be returned early, bypassing the actual save operation. If None, the save operation will proceed normally.

after_save​

def after_save(result: Union[dict, Exception], *args: Any,
               **kwargs: Any) -> Optional[dict]

[view_source]

Optional callback called after the resource is saved.

Override this method to add custom logic after saving.

Arguments:

  • result - The result from the save operation (dict on success, Exception on failure)
  • *args - Positional arguments that were passed to the save operation
  • **kwargs - Keyword arguments that were passed to the save operation

Returns:

  • Optional[dict] - If not None, this result will be returned instead of the original result. If None, the original result will be returned.

build_save_payload​

def build_save_payload(**kwargs: Any) -> dict

[view_source]

Build the payload for the save action.

save​

@with_hooks
def save(*args: Any, **kwargs: Any) -> "BaseResource"

[view_source]

Save the resource with attribute shortcuts.

This generic implementation provides consistent save behavior across all resources:

  • Supports attribute shortcuts: resource.save(name="new_name", description="...")
  • Lets the backend handle validation (name uniqueness, ID existence, etc.)
  • If the resource has an ID, it will be updated, otherwise it will be created.

Arguments:

  • *args - Positional arguments (not used, but kept for compatibility)
  • id - Optional[str] - Set resource ID before saving
  • name - Optional[str] - Set resource name before saving
  • description - Optional[str] - Set resource description before saving
  • **kwargs - Other attributes to set before saving

Returns:

  • BaseResource - The saved resource instance

Raises:

  • ResourceError - If the resource has been deleted — the same type every other deleted-save guard raises, so one except clause covers BaseResource.save, Agent.save and File.save. Backend validation errors as appropriate

clone​

@with_hooks
def clone(**kwargs: Any) -> "BaseResource"

[view_source]

Clone the resource and return a copy with id=None.

This generic implementation provides consistent clone behavior across all resources:

  • Creates deep copy of the resource
  • Resets id=None and _saved_state=None
  • Supports attribute shortcuts: resource.clone(name="new_name", version="2.0")
  • Uses hook system for subclass-specific logic (status handling, etc.)

Arguments:

  • name - Optional[str] - Set name on cloned resource
  • description - Optional[str] - Set description on cloned resource
  • **kwargs - Other attributes to set on cloned resource

Returns:

  • BaseResource - New resource instance with id=None

__repr__​

def __repr__() -> str

[view_source]

Return a string representation using path > id priority.

__str__​

def __str__() -> str

[view_source]

Return string representation of the resource.

encoded_id​

@property
def encoded_id() -> str

[view_source]

Get the URL-encoded version of the resource ID.

Returns:

The URL-encoded resource ID, or empty string if no ID exists

BaseParams Objects​

class BaseParams(TypedDict)

[view_source]

Base class for parameters that include API key and resource path.

Attributes:

  • api_key - Accepted for backward compatibility and ignored. Authentication always comes from the Aixplain context (Client._auth_headers), which overrides any per-call value. It is stripped from every request so it can never reach a model input body or a supplier's prompt logs. Configure credentials via Aixplain(api_key=...) or TEAM_API_KEY / AIXPLAIN_API_KEY instead.
  • resource_path - Custom resource path for actions (optional). Consumed by the URL builders; never forwarded to the backend body or to requests.

BaseSearchParams Objects​

class BaseSearchParams(BaseParams)

[view_source]

Base class for all search parameters.

Attributes:

  • query - str: The query string.
  • ownership - Tuple[OwnershipType, List[OwnershipType]]: The ownership type.
  • sort_by - SortBy: The attribute to sort by.
  • sort_order - SortOrder: The order to sort by.
  • page_number - int: The page number.
  • page_size - int: The page size.
  • resource_path - str: Optional custom resource path to override RESOURCE_PATH.
  • paginate_items_key - str: Optional key name for items in paginated response (overrides PAGINATE_ITEMS_KEY).
  • strict - bool: Whether a record that cannot be deserialized raises instead of being skipped (default: skip, correcting Page.total). Overrides PAGINATE_STRICT.

BaseGetParams Objects​

class BaseGetParams(BaseParams)

[view_source]

Base class for all get parameters.

Attributes:

  • host - The host URL for the request (optional).

BaseDeleteParams Objects​

class BaseDeleteParams(BaseParams)

[view_source]

Base class for all delete parameters.

BaseRunParams Objects​

class BaseRunParams(BaseParams)

[view_source]

Base class for all run parameters.

Attributes:

  • timeout - Maximum time in seconds to wait for completion.
  • wait_time - Initial interval in seconds between poll attempts.
  • run_retries - Extra submission attempts after the first failure (total attempts = 1 + run_retries). Covers only the POST that starts the
  • run - a failure while polling an already-running job is not retried, because re-submitting would start and bill a second execution.
  • run_retry_wait - Seconds to wait between retry attempts (default 1.0).

Notes:

session_id and agent_name are header-only run keys handled by every runnable (_headers_for_run → x-session-id / x-agent, stripped from the body by _RUN_CONTROL_KEYS). They are declared on ModelRunParams rather than here: AgentRunParams deliberately has no session_id — agent runs join a conversation via session=, and a look-alike key that only set a header would be a footgun.

BaseResult Objects​

@dataclass_json

@dataclass
class BaseResult()

[view_source]

Abstract base class for running results.

This class provides a minimal interface that concrete result classes should implement. Subclasses are responsible for defining their own fields and handling their specific data structures.

Result Objects​

@dataclass_json

@dataclass(repr=False)
class Result(BaseResult)

[view_source]

Default implementation of running results with common fields.

__getattr__​

def __getattr__(name: str) -> Any

[view_source]

Allow access to any field from the raw response data.

__repr__​

def __repr__() -> str

[view_source]

Return a formatted string representation with truncated data.

DeleteResult Objects​

@dataclass_json

@dataclass
class DeleteResult(Result)

[view_source]

Result for delete operations.

Page Objects​

class Page(Generic[ResourceT])

[view_source]

A paginated page of resources.

Attributes:

  • results - The list of resources in this page.
  • page_number - Current page number (0-indexed).
  • page_total - Total number of pages.
  • total - Total number of resources across all pages. Never counts a record that was returned by the API but skipped by this page.
  • skipped - Number of records the API returned for this page that could not be deserialized and were skipped. Always 0 under strict=True, where such records raise instead.

__init__​

def __init__(results: List[ResourceT],
             page_number: int,
             page_total: int,
             total: int,
             skipped: int = 0)

[view_source]

Initialize a Page instance.

Arguments:

  • results - List of resource instances in this page
  • page_number - Current page number (0-indexed)
  • page_total - Total number of pages
  • total - Total number of resources across all pages
  • skipped - Number of returned records that could not be deserialized

__repr__​

def __repr__() -> str

[view_source]

Return JSON representation of the page.

__iter__​

def __iter__()

[view_source]

Iterate over the results in this page.

__getitem__​

def __getitem__(key: str)

[view_source]

Allow dictionary-like access to page attributes.

SearchResourceMixin Objects​

class SearchResourceMixin(BaseMixin, Generic[SearchParamsT, ResourceT])

[view_source]

Mixin for listing resources with pagination and search functionality.

Attributes:

  • PAGINATE_PATH - str: The path for pagination.
  • PAGINATE_METHOD - str: The method for pagination.
  • PAGINATE_ITEMS_KEY - str: The key for the response.
  • PAGINATE_TOTAL_KEY - str: The key for the total number of resources.
  • PAGINATE_PAGE_TOTAL_KEY - str: The key for the total number of pages.
  • PAGINATE_DEFAULT_PAGE_NUMBER - int: The default page number.
  • PAGINATE_DEFAULT_PAGE_SIZE - int: The default page size.
  • PAGINATE_STRICT - bool: Whether a record that cannot be deserialized raises instead of being skipped and subtracted from Page.total. Defaults to False: discovery listings must survive a record the client cannot model yet (a new backend enum value, say). A per-call strict= keyword takes precedence over this class-level default.

PAGINATE_ITEMS_KEY​

Default to match backend

@classmethod
def search(cls: type, **kwargs: Unpack[SearchParamsT]) -> Page[ResourceT]

[view_source]

Search resources across the first n pages with optional filtering.

A record the API returns but this client cannot deserialize is logged and skipped rather than failing the whole listing; Page.total is reduced by the number skipped and Page.skipped reports the count, so page.total never counts a record the page does not contain. Pass strict=True (or set PAGINATE_STRICT) to raise ResourceError on such a record instead.

A response that carries no readable item list at all always raises: there is nothing to count, and an empty page would be indistinguishable from a search that genuinely matched nothing.

Arguments:

  • kwargs - The keyword arguments.

Returns:

  • Page[ResourceT] - Page of BaseResource instances

Raises:

  • ResourceError - If the response envelope carries no item list, or if a returned record cannot be deserialized and strict is on.

GetResourceMixin Objects​

class GetResourceMixin(BaseMixin, Generic[GetParamsT, ResourceT])

[view_source]

Mixin for getting a resource.

get​

@classmethod
def get(cls: type,
        id: Any,
        host: Optional[str] = None,
        **kwargs: Unpack[GetParamsT]) -> ResourceT

[view_source]

Retrieve a single resource by its ID (or other get parameters).

Arguments:

  • id - Any: The ID of the resource to get.
  • host - str, optional: The host parameter to pass to the backend (default: None).
  • kwargs - Get parameters to pass to the request.

Returns:

  • BaseResource - Instance of the BaseResource class.

Raises:

  • ValueError - If 'RESOURCE_PATH' is not defined by the subclass.

DeleteResourceMixin Objects​

class DeleteResourceMixin(BaseMixin, Generic[DeleteParamsT, DeleteResultT])

[view_source]

Mixin for deleting a resource.

DELETE_RESPONSE_CLASS​

Default response class

build_delete_payload​

def build_delete_payload(**kwargs: Unpack[DeleteParamsT]) -> dict

[view_source]

Build the payload for the delete action.

This method can be overridden by subclasses to provide custom payload construction for delete operations.

build_delete_url​

def build_delete_url(**kwargs: Unpack[DeleteParamsT]) -> str

[view_source]

Build the URL for the delete action.

This method can be overridden by subclasses to provide custom URL construction. The default implementation uses the resource path with the resource ID.

Returns:

  • str - The URL to use for the delete action

handle_delete_response​

def handle_delete_response(response: Any,
                           **kwargs: Unpack[DeleteParamsT]) -> DeleteResultT

[view_source]

Handle the response from a delete request.

This method can be overridden by subclasses to handle different response patterns. The default implementation creates a simple success response.

Arguments:

  • response - The raw response from the API (may be Response object or dict)
  • **kwargs - Delete parameters

Returns:

DeleteResult instance from the configured response class

before_delete​

def before_delete(*args: Any,
                  **kwargs: Unpack[DeleteParamsT]) -> Optional[DeleteResultT]

[view_source]

Optional callback called before the resource is deleted.

Override this method to add custom logic before deleting.

Arguments:

  • *args - Positional arguments passed to the delete operation
  • **kwargs - Keyword arguments passed to the delete operation

Returns:

  • Optional[DeleteResultT] - If not None, this result will be returned early, bypassing the actual delete operation. If None, the delete operation will proceed normally.

after_delete​

def after_delete(result: Union[DeleteResultT, Exception], *args: Any,
                 **kwargs: Unpack[DeleteParamsT]) -> Optional[DeleteResultT]

[view_source]

Optional callback called after the resource is deleted.

Override this method to add custom logic after deleting.

Arguments:

  • result - The result from the delete operation (DeleteResultT on success, Exception on failure)
  • *args - Positional arguments that were passed to the delete operation
  • **kwargs - Keyword arguments that were passed to the delete operation

Returns:

  • Optional[DeleteResultT] - If not None, this result will be returned instead of the original result. If None, the original result will be returned.

delete​

@with_hooks
def delete(*args: Any, **kwargs: Unpack[DeleteParamsT]) -> DeleteResultT

[view_source]

Delete a resource.

Returns:

  • DeleteResultT - The result of the delete operation

mark_as_deleted​

def mark_as_deleted() -> None

[view_source]

Mark the resource as deleted by clearing its ID and setting deletion flag.

RunnableResourceMixin Objects​

class RunnableResourceMixin(BaseMixin, Generic[RunParamsT, ResultT])

[view_source]

Mixin for runnable resources.

RESPONSE_CLASS​

Default response class

build_run_payload​

def build_run_payload(**kwargs: Unpack[RunParamsT]) -> dict

[view_source]

Build the payload for the run action.

This method automatically handles dataclass serialization if the run parameters are dataclasses with @dataclass_json decorator.

build_run_url​

def build_run_url(**kwargs: Unpack[RunParamsT]) -> str

[view_source]

Build the URL for the run action.

This method can be overridden by subclasses to provide custom URL construction. The default implementation uses the resource path with the run action.

Returns:

  • str - The URL to use for the run action

handle_run_response​

def handle_run_response(response: dict,
                        **kwargs: Unpack[RunParamsT]) -> ResultT

[view_source]

Handle the response from a run request.

This method can be overridden by subclasses to handle different response patterns. The default implementation assumes a polling URL in the 'data' field.

Arguments:

  • response - The raw response from the API
  • **kwargs - Run parameters

Returns:

Response instance from the configured response class

before_run​

def before_run(*args: Any, **kwargs: Unpack[RunParamsT]) -> Optional[ResultT]

[view_source]

Optional callback called before the resource is run.

Override this method to add custom logic before running.

Arguments:

  • *args - Positional arguments passed to the run operation
  • **kwargs - Keyword arguments passed to the run operation

Returns:

  • Optional[ResultT] - If not None, this result will be returned early, bypassing the actual run operation. If None, the run operation will proceed normally.

after_run​

def after_run(result: Union[ResultT, BaseException], *args: Any,
              **kwargs: Unpack[RunParamsT]) -> Optional[ResultT]

[view_source]

Optional callback called after the resource is run.

Override this method to add custom logic after running.

Arguments:

  • result - The result from the run operation: ResultT on success, or the raised BaseException on failure. This includes KeyboardInterrupt and SystemExit, so a hook must test isinstance(result, BaseException) rather than Exception before treating result as a result.
  • *args - Positional arguments that were passed to the run operation
  • **kwargs - Keyword arguments that were passed to the run operation

Returns:

  • Optional[ResultT] - If not None, this result will be returned instead of the original result. If None, the original result will be returned. On the failure path the return value is ignored and the exception propagates.

run​

def run(*args: Any, **kwargs: Unpack[RunParamsT]) -> ResultT

[view_source]

Run the resource synchronously with automatic polling.

Arguments:

  • *args - Positional arguments (converted to kwargs by subclasses)
  • **kwargs - Run parameters including timeout, wait_time, run_retries, run_retry_wait

Returns:

Response instance from the configured response class

Notes:

before_run runs once per run() call. Retries (if configured) re-submit the POST only, without invoking before_run again; polling happens outside the retry boundary, so a failure while watching an already-running job raises instead of starting — and billing — a second execution.

after_run runs exactly once per run() call on every exit path from the run itself -- success, submission failure, polling

  • failure - so whatever before_run started is always torn down. On the failure path its return value is ignored and the original exception propagates. The one path it does not cover is before_run raising: nothing has been set up yet, so a hook must not leave a resource live behind a raise of its own.

run_async​

def run_async(**kwargs: Unpack[RunParamsT]) -> ResultT

[view_source]

Run the resource asynchronously.

Arguments:

  • **kwargs - Run parameters specific to the resource type, including optional run_retries and run_retry_wait for failed POST submissions. An immediate FAILED response is a business outcome and is never retried.

Returns:

Response instance from the configured RESPONSE_CLASS

poll​

def poll(poll_url: str, timeout: Optional[float] = None) -> ResultT

[view_source]

Poll for the result of an asynchronous operation.

Arguments:

  • poll_url - URL to poll for results
  • timeout - Optional upper bound, in seconds, on the wall clock this single poll may consume -- normally the budget sync_poll has left. It is divided by the number of transport-level attempts before being applied as a read timeout, because urllib3 retries a timed-out GET underneath requests. When omitted the client's own default read timeout applies, which is long enough for one hung poll to consume an entire poll budget (BUG-1097).

Returns:

Response instance from the configured RESPONSE_CLASS

Raises:

  • UntrustedURLError - If poll_url resolves to a host outside the trusted set
  • APIError - If the polling request fails
  • OperationFailedError - If the operation has failed

on_poll​

def on_poll(response: ResultT, **kwargs: Unpack[RunParamsT]) -> None

[view_source]

Hook called after each successful poll with the poll response.

Override this method in subclasses to handle poll responses, such as displaying progress updates or logging status changes.

Arguments:

  • response - The response from the poll operation
  • **kwargs - Run parameters including show_progress, timeout, wait_time, etc.

sync_poll​

def sync_poll(poll_url: str, **kwargs: Unpack[RunParamsT]) -> ResultT

[view_source]

Keep polling until an asynchronous operation is complete.

Arguments:

  • poll_url - URL to poll for results. Must be the exact URL returned by run_async (e.g. result.url). Do not use a constructed /sdk/runs/{id} URL — that endpoint is not supported for agent runs.
  • **kwargs - Run parameters including timeout and wait_time

Returns:

Response instance from the configured RESPONSE_CLASS

Raises:

  • TimeoutError - If the operation exceeds the timeout duration. When the budget expired on a poll that also failed (a 401 on the last attempt, say), that failure is chained as __cause__ rather than being flattened into "Operation timed out".