Library Reference¶

This module provides the low-level client interface for interacting with the API of the Aignostics Platform.

The primary class in this module is the Client class, serving as the entry point for authenticated API operations. Login and token management are handled automatically.

Further operations are encapsulated in the Service class, which provides methods for manual login, logout and getting information about the authenticated user.

Higher level abstractions are provided in the application module.

exception aignostics.platform.ApiException(status=None, reason=None, http_resp=None, *, body: str | None = None, data: Any | None = None)¶

__str__()¶: Custom error messages for exception

aignostics.platform.Application¶: alias of ApplicationReadResponse

aignostics.platform.ApplicationSummary¶: alias of ApplicationReadShortResponse

aignostics.platform.ApplicationVersion¶: alias of VersionReadResponse

class aignostics.platform.Client(cache_token: bool = True)¶

Main client for interacting with the Aignostics Platform API.

Provides access to platform resources like applications, versions, and runs.
Handles authentication and API client configuration.
Retries on network and server errors for specific operations.
Caches operation results for specific operations.

application(application_id: str, nocache: bool = False) → ApplicationReadResponse¶

Find application by id.

Retries on network and server errors.

Parameters:

application_id (str) – The ID of the application.
nocache (bool) – If True, skip reading from cache and fetch fresh data from the API. The fresh result will still be cached for subsequent calls. Defaults to False.

Raises:

NotFoundException – If the application with the given ID is not found.
aignx.codegen.exceptions.ApiException – If the API call fails.

Returns:

The application object.

Return type:

Application

application_version(application_id: str, version_number: str | None = None, nocache: bool = False) → VersionReadResponse¶

Find application version by id.

Retries on network and server errors.

Parameters:

application_id (str) – The ID of the application.
version_number (str | None) – The version number of the application. If None, the latest version will be retrieved.
nocache (bool) – If True, skip reading from cache and fetch fresh data from the API. The fresh result will still be cached for subsequent calls. Defaults to False.

Raises:

NotFoundException – If the application with the given ID and version number is not found.
ValueError – If the version is not valid semver.
aignx.codegen.exceptions.ApiException – If the API call fails.

Returns:

The application version object.

Return type:

ApplicationVersion

static get_api_client(cache_token: bool = True) → PublicApi¶

Create and configure an authenticated API client.

API client instances are shared across all Client instances for efficient connection reuse. Two separate instances are maintained: one for cached tokens and one for uncached tokens.

Parameters:: cache_token (bool) – If True, caches the authentication token. Defaults to True.
Returns:: Configured API client with authentication token.
Return type:: PublicApi
Raises:: RuntimeError – If authentication fails.

me(nocache: bool = False) → MeReadResponse¶

Retrieves info about the current user and their organisation.

Retries on network and server errors.

Note: - We are not using urllib3s retry class as it does not support fine grained definition when to retry,

exponential backoff with jitter, logging before retry, and is difficult to configure.

Parameters:: nocache (bool) – If True, skip reading from cache and fetch fresh data from the API. The fresh result will still be cached for subsequent calls. Defaults to False.
Returns:: User and organization information.
Return type:: Me
Raises:: aignx.codegen.exceptions.ApiException – If the API call fails.

run(run_id: str) → Run¶

Finds run by id.

Parameters:: run_id (str) – The ID of the application run.
Returns:: The run object.
Return type:: Run

exception aignostics.platform.ForbiddenException(status=None, reason=None, http_resp=None, *, body: str | None = None, data: Any | None = None)¶

aignostics.platform.InputArtifact¶: alias of InputArtifactCreationRequest

aignostics.platform.InputArtifactData¶: alias of InputArtifact

aignostics.platform.InputItem¶: alias of ItemCreationRequest

class aignostics.platform.ItemOutput(*values)¶

classmethod from_json(json_str: str) → Self¶: Create an instance of ItemOutput from a JSON string

aignostics.platform.ItemResult¶: alias of ItemResultReadResponse

class aignostics.platform.ItemState(*values)¶

classmethod from_json(json_str: str) → Self¶: Create an instance of ItemState from a JSON string

class aignostics.platform.ItemTerminationReason(*values)¶

classmethod from_json(json_str: str) → Self¶: Create an instance of ItemTerminationReason from a JSON string

aignostics.platform.Me¶: alias of MeReadResponse

exception aignostics.platform.NotFoundException(status=None, reason=None, http_resp=None, *, body: str | None = None, data: Any | None = None)¶

aignostics.platform.Organization¶: alias of OrganizationReadResponse

aignostics.platform.OutputArtifactData¶: alias of OutputArtifact

aignostics.platform.OutputArtifactElement¶: alias of OutputArtifactResultReadResponse

pydantic model aignostics.platform.PipelineConfig¶

Pipeline configuration for dynamic orchestration.

Fields:

cpu (aignostics.platform._sdk_metadata.CPUConfig)
gpu (aignostics.platform._sdk_metadata.GPUConfig)
node_acquisition_timeout_minutes (int)

field cpu: CPUConfig [Optional]¶: CPU resource configuration

field gpu: GPUConfig [Optional]¶: GPU resource configuration

field node_acquisition_timeout_minutes: int = 30¶

Timeout for acquiring compute nodes in minutes (1-3600)

Constraints:

ge = 1
le = 3600

class aignostics.platform.Run(api: PublicApi, run_id: str)¶

Represents a single application run.

Provides operations to check status, retrieve results, and download artifacts.

__str__() → str¶

Returns a string representation of the application run.

The string includes run ID, status, and item statistics.

Returns:: String representation of the application run.
Return type:: str

cancel() → None¶

Cancels the application run.

Raises:: Exception – If the API request fails.

delete() → None¶

Delete the application run.

Raises:: Exception – If the API request fails.

details(nocache: bool = False, hide_platform_queue_position: bool = False) → RunReadResponse¶

Retrieves the current status of the application run.

Retries on network and server errors. Additionally retries on NotFoundException for up to 5 seconds to handle read replica lag.

Parameters:

nocache (bool) – If True, skip reading from cache and fetch fresh data from the API. The fresh result will still be cached for subsequent calls. Defaults to False.
hide_platform_queue_position (bool) – If True, hides the platform queue position in the returned run data. Defaults to False.

Returns:

The run data.

Return type:

RunData

Raises:

NotFoundException – If the run is not found after retries.
Exception – If the API request fails.

download_to_folder(download_base: Path | str, checksum_attribute_key: str = 'checksum_base64_crc32c', sleep_interval: int = 5, timeout_seconds: int | None = None, print_status: bool = True) → None¶

Downloads all result artifacts to a folder.

Monitors run progress and downloads results as they become available.

Parameters:

download_base (Path | str) – Base directory to download results to.
checksum_attribute_key (str) – The key used to validate the checksum of the output artifacts.
sleep_interval (int) – Time in seconds to wait between checks for new results.
timeout_seconds (int | None) – Optional timeout in seconds for the entire download operation.
print_status (bool) – If True, prints status updates to the console, otherwise just logs.

Raises:

ValueError – If the provided path is not a directory.
DownloadTimeoutError – If the timeout is exceeded while waiting for the run to terminate.
RuntimeError – If downloads or API requests fail.

static ensure_artifacts_downloaded(base_folder: Path, item: ItemResultReadResponse, checksum_attribute_key: str = 'checksum_base64_crc32c', print_status: bool = True) → None¶

Ensures all artifacts for an item are downloaded.

Downloads missing or partially downloaded artifacts and verifies their integrity.

Parameters:

base_folder (Path) – Base directory to download artifacts to.
item (ItemResultReadResponse) – The item result containing the artifacts to download.
checksum_attribute_key (str) – The key used to validate the checksum of the output artifacts.
print_status (bool) – If True, prints status updates to the console, otherwise just logs.

Raises:

ValueError – If checksums don’t match.
Exception – If downloads fail.

classmethod for_run_id(run_id: str, cache_token: bool = True) → Run¶

Creates an Run instance for an existing run.

Parameters:

run_id (str) – The ID of the application run.
cache_token (bool) – Whether to cache the API token.

Returns:

The initialized Run instance.

Return type:

Run

results(nocache: bool = False, item_ids: list[str] | None = None, external_ids: list[str] | None = None) → Iterator[ItemResultReadResponse]¶

Retrieves the results of all items in the run.

Retries on network and server errors.

Parameters:

nocache (bool) – If True, skip reading from cache and fetch fresh data from the API. The fresh result will still be cached for subsequent calls. Defaults to False.
item_ids (list[str] | None) – Optional list of item IDs to filter results by.
external_ids (list[str] | None) – Optional list of external IDs to filter results by.

Returns:

An iterator over item results.

Return type:

Iterator[ItemResultData]

Raises:

Exception – If the API request fails.

update_custom_metadata(custom_metadata: dict[str, Any]) → None¶

Update custom metadata for this application run.

Parameters:: custom_metadata (dict[str, Any]) – The new custom metadata to attach to the run.
Raises:: Exception – If the API request fails.

update_item_custom_metadata(external_id: str, custom_metadata: dict[str, Any]) → None¶

Update custom metadata for an item in this application run.

Parameters:

external_id (str) – The external ID of the item.
custom_metadata (dict[str, Any]) – The new custom metadata to attach to the item.

Raises:

Exception – If the API request fails.

aignostics.platform.RunData¶: alias of RunReadResponse

pydantic model aignostics.platform.RunItemStatistics¶

RunItemStatistics

Config:

populate_by_name: bool = True
validate_assignment: bool = True
protected_namespaces: tuple = ()
validate_by_alias: bool = True
validate_by_name: bool = True

Fields:

item_count (int)
item_pending_count (int)
item_processing_count (int)
item_skipped_count (int)
item_succeeded_count (int)
item_system_error_count (int)
item_user_error_count (int)

field item_count: StrictInt [Required]¶

Total number of the items in the run

Constraints:

strict = True

field item_pending_count: StrictInt [Required]¶

The number of items in PENDING state

Constraints:

strict = True

field item_processing_count: StrictInt [Required]¶

The number of items in PROCESSING state

Constraints:

strict = True

field item_skipped_count: StrictInt [Required]¶

The number of items in TERMINATED state, and the item termination reason is SKIPPED

Constraints:

strict = True

field item_succeeded_count: StrictInt [Required]¶

The number of items in TERMINATED state, and the item termination reason is SUCCEEDED

Constraints:

strict = True

field item_system_error_count: StrictInt [Required]¶

The number of items in TERMINATED state, and the item termination reason is SYSTEM_ERROR

Constraints:

strict = True

field item_user_error_count: StrictInt [Required]¶

The number of items in TERMINATED state, and the item termination reason is USER_ERROR

Constraints:

strict = True

classmethod from_dict(obj: Dict[str, Any] | None) → Self | None¶: Create an instance of RunItemStatistics from a dict

classmethod from_json(json_str: str) → Self | None¶: Create an instance of RunItemStatistics from a JSON string

to_dict() → Dict[str, Any]¶

Return the dictionary representation of the model using alias.

This has the following differences from calling pydantic’s self.model_dump(by_alias=True):

None is only added to the output dict for nullable fields that were set at model initialization. Other fields with value None are ignored.

to_json() → str¶: Returns the JSON representation of the model using alias

to_str() → str¶: Returns the string representation of the model using alias

class aignostics.platform.RunOutput(*values)¶

classmethod from_json(json_str: str) → Self¶: Create an instance of RunOutput from a JSON string

pydantic model aignostics.platform.RunSdkMetadata¶

Complete Run SDK metadata schema.

This model defines the structure and validation rules for SDK metadata that is attached to application runs. It includes information about: - SDK version and timestamps - User information (when available) - CI/CD environment context (GitHub Actions, pytest) - Workflow control flags - Scheduling information - Optional user note

Config:

extra: str = forbid

Fields:

ci (aignostics.platform._sdk_metadata.CIMetadata | None)
created_at (str)
note (str | None)
pipeline (aignostics.platform._sdk_metadata.PipelineConfig | None)
scheduling (aignostics.platform._sdk_metadata.SchedulingMetadata | None)
schema_version (str)
submission (aignostics.platform._sdk_metadata.SubmissionMetadata)
tags (set[str] | None)
updated_at (str)
user (aignostics.platform._sdk_metadata.UserMetadata | None)
user_agent (str)
workflow (aignostics.platform._sdk_metadata.WorkflowMetadata | None)

Validators:

validate_validation_case » tags

field ci: CIMetadata | None = None¶: CI/CD environment metadata

field created_at: str [Required]¶: ISO 8601 timestamp when the metadata was first created

field note: str | None = None¶: Optional user note for the run

field pipeline: PipelineConfig | None = None¶: Pipeline orchestration configuration

field scheduling: SchedulingMetadata | None = None¶: Scheduling information

field schema_version: str [Required]¶

Schema version for this metadata format

Constraints:

pattern = ^d+.d+.d+-?.*$

field submission: SubmissionMetadata [Required]¶: Submission context metadata

field tags: set[str] | None = None¶

Optional list of tags associated with the run

Validated by:

validate_validation_case

field updated_at: str [Required]¶: ISO 8601 timestamp when the metadata was last updated

field user: UserMetadata | None = None¶: User information (when authenticated)

field user_agent: str [Required]¶: User agent string for the SDK client

field workflow: WorkflowMetadata | None = None¶: Workflow control flags

validator validate_validation_case » tags¶

class aignostics.platform.RunState(*values)¶

classmethod from_json(json_str: str) → Self¶: Create an instance of RunState from a JSON string

class aignostics.platform.RunTerminationReason(*values)¶

classmethod from_json(json_str: str) → Self¶: Create an instance of RunTerminationReason from a JSON string

pydantic model aignostics.platform.SchedulingMetadata¶

Scheduling metadata for run execution.

Fields:

deadline (str | None)
due_date (str | None)

field deadline: str | None = None¶: Hard deadline (ISO 8601). Run may be aborted if processing exceeds this time.

field due_date: str | None = None¶: Requested completion time (ISO 8601). Scheduler will try to complete before this time.

class aignostics.platform.Service¶

Service of the application module.

static get_user_info(relogin: bool = False) → UserInfo¶

Get user information from authentication token.

Parameters:: relogin (bool) – If True, forces a re-login even if a token is cached.
Returns:: User information if successfully authenticated, None if login failed.
Return type:: UserInfo | None
Raises:: RuntimeError – If the token cannot be verified or decoded.

async health() → Health¶

Determine health of this service.

Returns:: The health of the service.
Return type:: Health

async info(mask_secrets: bool = True) → dict[str, Any]¶

Determine info of this service.

Parameters:: mask_secrets (bool) – Whether to mask sensitive information in the output.
Returns:: The info of this service.
Return type:: dict[str,Any]

static login(relogin: bool = False) → bool¶

Parameters:: relogin (bool) – If True, forces a re-login even if a token is cached.
Returns:: True if successfully logged in, False if login failed
Return type:: bool

static logout() → bool¶

Logout if authenticated.

Deletes the cached authentication token if existing.

Returns:: True if successfully logged out, False if not logged in.
Return type:: bool

pydantic settings aignostics.platform.Settings¶

Configuration settings for the Aignostics SDK.

This class handles configuration settings loaded from environment variables, configuration files, or default values. It manages authentication endpoints, client credentials, token storage, and other SDK behaviors.

Show JSON schema

{
   "title": "Settings",
   "description": "Configuration settings for the Aignostics SDK.\n\nThis class handles configuration settings loaded from environment variables,\nconfiguration files, or default values. It manages authentication endpoints,\nclient credentials, token storage, and other SDK behaviors.",
   "type": "object",
   "properties": {
      "client_id_device": {
         "anyOf": [
            {
               "format": "password",
               "type": "string",
               "writeOnly": true
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "OAuth Client ID Interactive",
         "title": "Client Id Device"
      },
      "api_root": {
         "default": "https://platform.aignostics.com",
         "description": "URL of the API root",
         "title": "Api Root",
         "type": "string"
      },
      "scope": {
         "default": "offline_access",
         "description": "OAuth scopes",
         "minLength": 3,
         "title": "Scope",
         "type": "string"
      },
      "audience": {
         "description": "OAuth audience claim",
         "maxLength": 100,
         "minLength": 10,
         "title": "Audience",
         "type": "string"
      },
      "authorization_base_url": {
         "description": "OAuth authorization endpoint URL",
         "minLength": 1,
         "title": "Authorization Base Url",
         "type": "string"
      },
      "token_url": {
         "description": "OAuth token endpoint URL",
         "title": "Token Url",
         "type": "string"
      },
      "redirect_uri": {
         "description": "OAuth redirect URI for authorization code flow",
         "title": "Redirect Uri",
         "type": "string"
      },
      "device_url": {
         "description": "OAuth device authorization endpoint URL",
         "title": "Device Url",
         "type": "string"
      },
      "jws_json_url": {
         "description": "JWS key set URL for token verification",
         "title": "Jws Json Url",
         "type": "string"
      },
      "client_id_interactive": {
         "description": "OAuth client ID for interactive flows",
         "title": "Client Id Interactive",
         "type": "string"
      },
      "refresh_token": {
         "anyOf": [
            {
               "format": "password",
               "maxLength": 1000,
               "minLength": 10,
               "type": "string",
               "writeOnly": true
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Refresh token for OAuth authentication",
         "title": "Refresh Token"
      },
      "cache_dir": {
         "default": "/home/docs/.cache/aignostics",
         "title": "Cache Dir",
         "type": "string"
      },
      "health_timeout": {
         "default": 30.0,
         "description": "Timeout for health checks",
         "maximum": 300.0,
         "minimum": 0.1,
         "title": "Health Timeout",
         "type": "number"
      },
      "auth_jwk_set_cache_ttl": {
         "default": 86400,
         "description": "Time-to-live for JWK set cache (in seconds)",
         "maximum": 604800,
         "minimum": 0,
         "title": "Auth Jwk Set Cache Ttl",
         "type": "integer"
      },
      "auth_timeout": {
         "default": 30.0,
         "description": "Timeout for authentication requests",
         "maximum": 300.0,
         "minimum": 0.1,
         "title": "Auth Timeout",
         "type": "number"
      },
      "auth_retry_attempts": {
         "default": 4,
         "description": "Number of retry attempts for authentication requests",
         "maximum": 10,
         "minimum": 0,
         "title": "Auth Retry Attempts",
         "type": "integer"
      },
      "auth_retry_wait_min": {
         "default": 0.1,
         "description": "Minimum wait time between retry attempts (in seconds)",
         "maximum": 600.0,
         "minimum": 0.0,
         "title": "Auth Retry Wait Min",
         "type": "number"
      },
      "auth_retry_wait_max": {
         "default": 60.0,
         "description": "Maximum wait time between retry attempts (in seconds)",
         "maximum": 600.0,
         "minimum": 0.0,
         "title": "Auth Retry Wait Max",
         "type": "number"
      },
      "me_timeout": {
         "default": 30.0,
         "description": "Timeout for me requests",
         "maximum": 300.0,
         "minimum": 0.1,
         "title": "Me Timeout",
         "type": "number"
      },
      "me_retry_attempts": {
         "default": 4,
         "description": "Number of retry attempts for me requests",
         "maximum": 10,
         "minimum": 0,
         "title": "Me Retry Attempts",
         "type": "integer"
      },
      "me_retry_wait_min": {
         "default": 0.1,
         "description": "Minimum wait time between retry attempts (in seconds)",
         "maximum": 600.0,
         "minimum": 0.0,
         "title": "Me Retry Wait Min",
         "type": "number"
      },
      "me_retry_wait_max": {
         "default": 60.0,
         "description": "Maximum wait time between retry attempts (in seconds)",
         "maximum": 600.0,
         "minimum": 0.0,
         "title": "Me Retry Wait Max",
         "type": "number"
      },
      "me_cache_ttl": {
         "default": 300,
         "description": "Time-to-live for me cache (in seconds)",
         "maximum": 604800,
         "minimum": 0,
         "title": "Me Cache Ttl",
         "type": "integer"
      },
      "application_timeout": {
         "default": 30.0,
         "description": "Timeout for application requests",
         "maximum": 300.0,
         "minimum": 0.1,
         "title": "Application Timeout",
         "type": "number"
      },
      "application_retry_attempts": {
         "default": 4,
         "description": "Number of retry attempts for application requests",
         "maximum": 10,
         "minimum": 0,
         "title": "Application Retry Attempts",
         "type": "integer"
      },
      "application_retry_wait_min": {
         "default": 0.1,
         "description": "Minimum wait time between retry attempts (in seconds)",
         "maximum": 600.0,
         "minimum": 0.0,
         "title": "Application Retry Wait Min",
         "type": "number"
      },
      "application_retry_wait_max": {
         "default": 60.0,
         "description": "Maximum wait time between retry attempts (in seconds)",
         "maximum": 600.0,
         "minimum": 0.0,
         "title": "Application Retry Wait Max",
         "type": "number"
      },
      "application_cache_ttl": {
         "default": 300,
         "description": "Time-to-live for application cache (in seconds)",
         "maximum": 604800,
         "minimum": 0,
         "title": "Application Cache Ttl",
         "type": "integer"
      },
      "application_version_timeout": {
         "default": 30.0,
         "description": "Timeout for application version requests",
         "maximum": 300.0,
         "minimum": 0.1,
         "title": "Application Version Timeout",
         "type": "number"
      },
      "application_version_retry_attempts": {
         "default": 4,
         "description": "Number of retry attempts for application version requests",
         "maximum": 10,
         "minimum": 0,
         "title": "Application Version Retry Attempts",
         "type": "integer"
      },
      "application_version_retry_wait_min": {
         "default": 0.1,
         "description": "Minimum wait time between retry attempts (in seconds)",
         "maximum": 600.0,
         "minimum": 0.0,
         "title": "Application Version Retry Wait Min",
         "type": "number"
      },
      "application_version_retry_wait_max": {
         "default": 60.0,
         "description": "Maximum wait time between retry attempts (in seconds)",
         "maximum": 600.0,
         "minimum": 0.0,
         "title": "Application Version Retry Wait Max",
         "type": "number"
      },
      "application_version_cache_ttl": {
         "default": 300,
         "description": "Time-to-live for application version cache (in seconds)",
         "maximum": 604800,
         "minimum": 0,
         "title": "Application Version Cache Ttl",
         "type": "integer"
      },
      "run_timeout": {
         "default": 30.0,
         "description": "Timeout for run requests",
         "maximum": 300.0,
         "minimum": 0.1,
         "title": "Run Timeout",
         "type": "number"
      },
      "run_retry_attempts": {
         "default": 4,
         "description": "Number of retry attempts for run requests",
         "maximum": 10,
         "minimum": 0,
         "title": "Run Retry Attempts",
         "type": "integer"
      },
      "run_retry_wait_min": {
         "default": 0.1,
         "description": "Minimum wait time between retry attempts (in seconds)",
         "maximum": 600.0,
         "minimum": 0.0,
         "title": "Run Retry Wait Min",
         "type": "number"
      },
      "run_retry_wait_max": {
         "default": 60.0,
         "description": "Maximum wait time between retry attempts (in seconds)",
         "maximum": 600.0,
         "minimum": 0.0,
         "title": "Run Retry Wait Max",
         "type": "number"
      },
      "run_cache_ttl": {
         "default": 15,
         "description": "Time-to-live for run cache (in seconds)",
         "maximum": 604800,
         "minimum": 0,
         "title": "Run Cache Ttl",
         "type": "integer"
      },
      "run_cancel_timeout": {
         "default": 30.0,
         "description": "Timeout for run cancel requests",
         "maximum": 300.0,
         "minimum": 0.1,
         "title": "Run Cancel Timeout",
         "type": "number"
      },
      "run_delete_timeout": {
         "default": 30.0,
         "description": "Timeout for run delete requests",
         "maximum": 300.0,
         "minimum": 0.1,
         "title": "Run Delete Timeout",
         "type": "number"
      },
      "run_submit_timeout": {
         "default": 30.0,
         "description": "Timeout for run submit requests",
         "maximum": 300.0,
         "minimum": 0.1,
         "title": "Run Submit Timeout",
         "type": "number"
      }
   },
   "required": [
      "audience",
      "authorization_base_url",
      "token_url",
      "redirect_uri",
      "device_url",
      "jws_json_url",
      "client_id_interactive"
   ]
}

Config:

extra: str = ignore
env_prefix: str = AIGNOSTICS_
env_file: tuple = (PosixPath(‘/home/docs/.aignostics/.env’), PosixPath(‘.env’))
env_file_encoding: str = utf-8

Fields:

api_root (str)
application_cache_ttl (int)
application_retry_attempts (int)
application_retry_wait_max (float)
application_retry_wait_min (float)
application_timeout (float)
application_version_cache_ttl (int)
application_version_retry_attempts (int)
application_version_retry_wait_max (float)
application_version_retry_wait_min (float)
application_version_timeout (float)
audience (str)
auth_jwk_set_cache_ttl (int)
auth_retry_attempts (int)
auth_retry_wait_max (float)
auth_retry_wait_min (float)
auth_timeout (float)
authorization_base_url (str)
cache_dir (str)
client_id_device (pydantic.types.SecretStr | None)
client_id_interactive (str)
device_url (str)
health_timeout (float)
jws_json_url (str)
me_cache_ttl (int)
me_retry_attempts (int)
me_retry_wait_max (float)
me_retry_wait_min (float)
me_timeout (float)
redirect_uri (str)
refresh_token (pydantic.types.SecretStr | None)
run_cache_ttl (int)
run_cancel_timeout (float)
run_delete_timeout (float)
run_retry_attempts (int)
run_retry_wait_max (float)
run_retry_wait_min (float)
run_submit_timeout (float)
run_timeout (float)
scope (str)
token_url (str)

Validators:

pre_init » all fields
validate_retry_wait_times » all fields

field api_root: Annotated[str, BeforeValidator(func=_validate_url, json_schema_input_type=PydanticUndefined), FieldInfo(annotation=NoneType, required=False, default='https://platform.aignostics.com', description='URL of the API root')] = 'https://platform.aignostics.com'¶

URL of the API root

Constraints:

func = <function _validate_url at 0x7de35ce6eae0>
json_schema_input_type = PydanticUndefined

Validated by:

pre_init
validate_retry_wait_times

field application_cache_ttl: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Time-to-live for application cache (in seconds)', metadata=[Ge(ge=0), Le(le=604800)])] = 300¶

Time-to-live for application cache (in seconds)

Constraints:

ge = 0
le = 604800

Validated by:

pre_init
validate_retry_wait_times

field application_retry_attempts: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Number of retry attempts for application requests', metadata=[Ge(ge=0), Le(le=10)])] = 4¶

Number of retry attempts for application requests

Constraints:

ge = 0
le = 10

Validated by:

pre_init
validate_retry_wait_times

field application_retry_wait_max: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Maximum wait time between retry attempts (in seconds)', metadata=[Ge(ge=0.0), Le(le=600.0)])] = 60.0¶

Maximum wait time between retry attempts (in seconds)

Constraints:

ge = 0.0
le = 600.0

Validated by:

pre_init
validate_retry_wait_times

field application_retry_wait_min: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Minimum wait time between retry attempts (in seconds)', metadata=[Ge(ge=0.0), Le(le=600.0)])] = 0.1¶

Minimum wait time between retry attempts (in seconds)

Constraints:

ge = 0.0
le = 600.0

Validated by:

pre_init
validate_retry_wait_times

field application_timeout: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Timeout for application requests', metadata=[Ge(ge=0.1), Le(le=300.0)])] = 30.0¶

Timeout for application requests

Constraints:

ge = 0.1
le = 300.0

Validated by:

pre_init
validate_retry_wait_times

field application_version_cache_ttl: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Time-to-live for application version cache (in seconds)', metadata=[Ge(ge=0), Le(le=604800)])] = 300¶

Time-to-live for application version cache (in seconds)

Constraints:

ge = 0
le = 604800

Validated by:

pre_init
validate_retry_wait_times

field application_version_retry_attempts: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Number of retry attempts for application version requests', metadata=[Ge(ge=0), Le(le=10)])] = 4¶

Number of retry attempts for application version requests

Constraints:

ge = 0
le = 10

Validated by:

pre_init
validate_retry_wait_times

field application_version_retry_wait_max: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Maximum wait time between retry attempts (in seconds)', metadata=[Ge(ge=0.0), Le(le=600.0)])] = 60.0¶

Maximum wait time between retry attempts (in seconds)

Constraints:

ge = 0.0
le = 600.0

Validated by:

pre_init
validate_retry_wait_times

field application_version_retry_wait_min: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Minimum wait time between retry attempts (in seconds)', metadata=[Ge(ge=0.0), Le(le=600.0)])] = 0.1¶

Minimum wait time between retry attempts (in seconds)

Constraints:

ge = 0.0
le = 600.0

Validated by:

pre_init
validate_retry_wait_times

field application_version_timeout: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Timeout for application version requests', metadata=[Ge(ge=0.1), Le(le=300.0)])] = 30.0¶

Timeout for application version requests

Constraints:

ge = 0.1
le = 300.0

Validated by:

pre_init
validate_retry_wait_times

field audience: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='OAuth audience claim', metadata=[MinLen(min_length=10), MaxLen(max_length=100)])] [Required]¶

OAuth audience claim

Constraints:

min_length = 10
max_length = 100

Validated by:

pre_init
validate_retry_wait_times

field auth_jwk_set_cache_ttl: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Time-to-live for JWK set cache (in seconds)', metadata=[Ge(ge=0), Le(le=604800)])] = 86400¶

Time-to-live for JWK set cache (in seconds)

Constraints:

ge = 0
le = 604800

Validated by:

pre_init
validate_retry_wait_times

field auth_retry_attempts: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Number of retry attempts for authentication requests', metadata=[Ge(ge=0), Le(le=10)])] = 4¶

Number of retry attempts for authentication requests

Constraints:

ge = 0
le = 10

Validated by:

pre_init
validate_retry_wait_times

field auth_retry_wait_max: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Maximum wait time between retry attempts (in seconds)', metadata=[Ge(ge=0.0), Le(le=600.0)])] = 60.0¶

Maximum wait time between retry attempts (in seconds)

Constraints:

ge = 0.0
le = 600.0

Validated by:

pre_init
validate_retry_wait_times

field auth_retry_wait_min: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Minimum wait time between retry attempts (in seconds)', metadata=[Ge(ge=0.0), Le(le=600.0)])] = 0.1¶

Minimum wait time between retry attempts (in seconds)

Constraints:

ge = 0.0
le = 600.0

Validated by:

pre_init
validate_retry_wait_times

field auth_timeout: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Timeout for authentication requests', metadata=[Ge(ge=0.1), Le(le=300.0)])] = 30.0¶

Timeout for authentication requests

Constraints:

ge = 0.1
le = 300.0

Validated by:

pre_init
validate_retry_wait_times

field authorization_base_url: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='OAuth authorization endpoint URL', metadata=[MinLen(min_length=1)])] [Required]¶

OAuth authorization endpoint URL

Constraints:

min_length = 1

Validated by:

pre_init
validate_retry_wait_times

field cache_dir: str = '/home/docs/.cache/aignostics'¶

Validated by:

pre_init
validate_retry_wait_times

field client_id_device: Annotated[SecretStr | None, PlainSerializer(func=serialize_sensitive_info, return_type=str, when_used=always), FieldInfo(annotation=NoneType, required=True, description='OAuth Client ID Interactive')] = None¶

OAuth Client ID Interactive

Constraints:

func = <function OpaqueSettings.serialize_sensitive_info at 0x7de35fd9f320>
return_type = <class ‘str’>
when_used = always

Validated by:

pre_init
validate_retry_wait_times

field client_id_interactive: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='OAuth client ID for interactive flows')] [Required]¶

OAuth client ID for interactive flows

Validated by:

pre_init
validate_retry_wait_times

field device_url: Annotated[str, BeforeValidator(func=_validate_url, json_schema_input_type=PydanticUndefined), FieldInfo(annotation=NoneType, required=True, description='OAuth device authorization endpoint URL')] [Required]¶

OAuth device authorization endpoint URL

Constraints:

func = <function _validate_url at 0x7de35ce6eae0>
json_schema_input_type = PydanticUndefined

Validated by:

pre_init
validate_retry_wait_times

field health_timeout: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Timeout for health checks', metadata=[Ge(ge=0.1), Le(le=300.0)])] = 30.0¶

Timeout for health checks

Constraints:

ge = 0.1
le = 300.0

Validated by:

pre_init
validate_retry_wait_times

field jws_json_url: Annotated[str, BeforeValidator(func=_validate_url, json_schema_input_type=PydanticUndefined), FieldInfo(annotation=NoneType, required=True, description='JWS key set URL for token verification')] [Required]¶

JWS key set URL for token verification

Constraints:

func = <function _validate_url at 0x7de35ce6eae0>
json_schema_input_type = PydanticUndefined

Validated by:

pre_init
validate_retry_wait_times

field me_cache_ttl: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Time-to-live for me cache (in seconds)', metadata=[Ge(ge=0), Le(le=604800)])] = 300¶

Time-to-live for me cache (in seconds)

Constraints:

ge = 0
le = 604800

Validated by:

pre_init
validate_retry_wait_times

field me_retry_attempts: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Number of retry attempts for me requests', metadata=[Ge(ge=0), Le(le=10)])] = 4¶

Number of retry attempts for me requests

Constraints:

ge = 0
le = 10

Validated by:

pre_init
validate_retry_wait_times

field me_retry_wait_max: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Maximum wait time between retry attempts (in seconds)', metadata=[Ge(ge=0.0), Le(le=600.0)])] = 60.0¶

Maximum wait time between retry attempts (in seconds)

Constraints:

ge = 0.0
le = 600.0

Validated by:

pre_init
validate_retry_wait_times

field me_retry_wait_min: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Minimum wait time between retry attempts (in seconds)', metadata=[Ge(ge=0.0), Le(le=600.0)])] = 0.1¶

Minimum wait time between retry attempts (in seconds)

Constraints:

ge = 0.0
le = 600.0

Validated by:

pre_init
validate_retry_wait_times

field me_timeout: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Timeout for me requests', metadata=[Ge(ge=0.1), Le(le=300.0)])] = 30.0¶

Timeout for me requests

Constraints:

ge = 0.1
le = 300.0

Validated by:

pre_init
validate_retry_wait_times

field redirect_uri: Annotated[str, BeforeValidator(func=_validate_url, json_schema_input_type=PydanticUndefined), FieldInfo(annotation=NoneType, required=True, description='OAuth redirect URI for authorization code flow')] [Required]¶

OAuth redirect URI for authorization code flow

Constraints:

func = <function _validate_url at 0x7de35ce6eae0>
json_schema_input_type = PydanticUndefined

Validated by:

pre_init
validate_retry_wait_times

field refresh_token: Annotated[SecretStr | None, PlainSerializer(func=serialize_sensitive_info, return_type=str, when_used=always), FieldInfo(annotation=NoneType, required=False, default=None, description='Refresh token for OAuth authentication', metadata=[MinLen(min_length=10), MaxLen(max_length=1000)])] = None¶

Refresh token for OAuth authentication

Constraints:

func = <function OpaqueSettings.serialize_sensitive_info at 0x7de35fd9f320>
return_type = <class ‘str’>
when_used = always
min_length = 10
max_length = 1000

Validated by:

pre_init
validate_retry_wait_times

field run_cache_ttl: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Time-to-live for run cache (in seconds)', metadata=[Ge(ge=0), Le(le=604800)])] = 15¶

Time-to-live for run cache (in seconds)

Constraints:

ge = 0
le = 604800

Validated by:

pre_init
validate_retry_wait_times

field run_cancel_timeout: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Timeout for run cancel requests', metadata=[Ge(ge=0.1), Le(le=300.0)])] = 30.0¶

Timeout for run cancel requests

Constraints:

ge = 0.1
le = 300.0

Validated by:

pre_init
validate_retry_wait_times

field run_delete_timeout: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Timeout for run delete requests', metadata=[Ge(ge=0.1), Le(le=300.0)])] = 30.0¶

Timeout for run delete requests

Constraints:

ge = 0.1
le = 300.0

Validated by:

pre_init
validate_retry_wait_times

field run_retry_attempts: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='Number of retry attempts for run requests', metadata=[Ge(ge=0), Le(le=10)])] = 4¶

Number of retry attempts for run requests

Constraints:

ge = 0
le = 10

Validated by:

pre_init
validate_retry_wait_times

field run_retry_wait_max: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Maximum wait time between retry attempts (in seconds)', metadata=[Ge(ge=0.0), Le(le=600.0)])] = 60.0¶

Maximum wait time between retry attempts (in seconds)

Constraints:

ge = 0.0
le = 600.0

Validated by:

pre_init
validate_retry_wait_times

field run_retry_wait_min: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Minimum wait time between retry attempts (in seconds)', metadata=[Ge(ge=0.0), Le(le=600.0)])] = 0.1¶

Minimum wait time between retry attempts (in seconds)

Constraints:

ge = 0.0
le = 600.0

Validated by:

pre_init
validate_retry_wait_times

field run_submit_timeout: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Timeout for run submit requests', metadata=[Ge(ge=0.1), Le(le=300.0)])] = 30.0¶

Timeout for run submit requests

Constraints:

ge = 0.1
le = 300.0

Validated by:

pre_init
validate_retry_wait_times

field run_timeout: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Timeout for run requests', metadata=[Ge(ge=0.1), Le(le=300.0)])] = 30.0¶

Timeout for run requests

Constraints:

ge = 0.1
le = 300.0

Validated by:

pre_init
validate_retry_wait_times

field scope: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='OAuth scopes', metadata=[MinLen(min_length=3)])] = 'offline_access'¶

OAuth scopes

Constraints:

min_length = 3

Validated by:

pre_init
validate_retry_wait_times

field token_url: Annotated[str, BeforeValidator(func=_validate_url, json_schema_input_type=PydanticUndefined), FieldInfo(annotation=NoneType, required=True, description='OAuth token endpoint URL')] [Required]¶

OAuth token endpoint URL

Constraints:

func = <function _validate_url at 0x7de35ce6eae0>
json_schema_input_type = PydanticUndefined

Validated by:

pre_init
validate_retry_wait_times

validator pre_init » all fields¶

Initialize auth-related fields based on the API root.

This validator sets the appropriate authentication URLs and parameters based on the target environment (production, staging, or development). If auth-related fields are already provided, they will not be overridden.

Parameters:: values – The input data dictionary to validate.
Returns:: The updated values dictionary with all environment-specific fields populated.
Raises:: ValueError – If the API root URL is not recognized and auth fields are missing.

serialize_token_file(token_file: Path, _info: FieldSerializationInfo) → str¶

validator validate_retry_wait_times » all fields¶

Validate that retry wait min is less or equal than retry wait max for all operations.

Returns:: The validated settings instance.
Return type:: Settings
Raises:: ValueError – If any operation’s retry_wait_min is greater than retry_wait_max.

property issuer: str¶

Get the issuer URL based on the authorization base root.

Extracts the scheme and domain from the authorization base URL to create a failsafe issuer URL in the format scheme://domain/

Returns:: Issuer URL in the format scheme://domain/
Return type:: str

property profile_edit_url: str¶

Get the profile edit URL based on the API root.

Returns:: Full URL to the profile editor page
Return type:: str

property scope_elements: list[str]¶

Get the OAuth scope elements as a list.

Splits the scope string by comma and strips whitespace from each element.

Returns:: List of individual scope elements.
Return type:: list[str]

property tenant_domain: str¶

Get the tenant domain from the authorization base URL.

Returns:: The domain part of the authorization base URL.
Return type:: str
Raises:: ValueError – If the authorization base URL is invalid or does not contain a netloc.

property token_file: Path¶

Get the path to the token file.

Returns:: The path to the file where the authentication token is stored.
Return type:: Path

pydantic model aignostics.platform.TokenInfo¶

Class to store token information.

Fields:

audience (list[str])
authorized_party (str)
expires_at (int)
issued_at (int)
issuer (str)
org_id (str)
role (str)
scope (list[str])

field audience: list[str] [Required]¶

field authorized_party: str [Required]¶

field expires_at: int [Required]¶

field issued_at: int [Required]¶

field issuer: str [Required]¶

field org_id: str [Required]¶

field role: str [Required]¶

field scope: list[str] [Required]¶

classmethod from_claims(claims: dict[str, Any]) → TokenInfo¶

Create TokenInfo from JWT claims.

Parameters:: claims – JWT token claims dictionary.
Returns:: Token information extracted from claims.
Return type:: TokenInfo

property expires_in: int¶

Calculate seconds until token expires.

Returns:: Number of seconds until the token expires. Negative if already expired.
Return type:: int

aignostics.platform.User¶: alias of UserReadResponse

pydantic model aignostics.platform.UserInfo¶

Class to store info about the user.

Fields:

organization (aignx.codegen.models.organization_read_response.OrganizationReadResponse)
role (str)
token (aignostics.platform._service.TokenInfo)
user (aignx.codegen.models.user_read_response.UserReadResponse)

field organization: OrganizationReadResponse [Required]¶

field role: str [Required]¶

field token: TokenInfo [Required]¶

field user: UserReadResponse [Required]¶

classmethod from_claims_and_me(claims: dict[str, Any], me: MeReadResponse) → UserInfo¶

Create UserInfo from JWT claims and optional auth0 userinfo.

Parameters:

claims (dict[str, Any]) – JWT token claims dictionary.
me (Me) – Info about calling user and their oganisation.

Returns:

User information extracted from claims.

Return type:

UserInfo

model_dump_json_secrets_masked() → str¶

Dump model to JSON with sensitive organization secrets masked.

Returns:

JSON representation with aignostics_bucket_hmac_access_key_id and: aignostics_bucket_hmac_secret_access_key masked.

Return type:

str

model_dump_secrets_masked() → dict[str, Any]¶

Dump model to dict with sensitive organization and user secrets masked.

Returns:: Dictionary representation with sensitive organization and user fields masked.
Return type:: dict[str, Any]

property is_internal_user: bool¶

Check if the user is an internal user.

Returns:: True if it is an internal user, False otherwise.
Return type:: bool

aignostics.platform.calculate_file_crc32c(file: Path) → str¶

Calculates the CRC32C checksum of a file.

Parameters:: file (Path) – Path to the file to calculate the checksum for.
Returns:: The CRC32C checksum in base64 encoding.
Return type:: str

aignostics.platform.download_file(signed_url: str, file_path: str, verify_checksum: str) → None¶

Downloads a file from a signed URL and verifies its integrity.

Parameters:

signed_url (str) – The signed URL to download the file from.
file_path (str) – The local path where the file should be saved.
verify_checksum (str) – The expected CRC32C checksum in base64 encoding.

Raises:

ValueError – If the downloaded file’s checksum doesn’t match the expected value.
requests.HTTPError – If the download request fails.

aignostics.platform.generate_signed_url(url: str, expires_seconds: int = 21600) → str¶

Generates a signed URL for a Google Cloud Storage object.

Parameters:

url (str) – The fully qualified bucket URL (e.g. gs://bucket/path/to/object).
expires_seconds (int) – The number of seconds the signed URL should be valid for.

Returns:

A signed URL that can be used to download the object.

Return type:

str

Raises:

ValueError – If the GS path is invalid or the blob doesn’t exist.

aignostics.platform.get_mime_type_for_artifact(artifact: OutputArtifact | InputArtifact | OutputArtifactResultReadResponse) → str¶

Get the MIME type for a given artifact.

Parameters:: artifact (OutputArtifact | InputArtifact | OutputArtifactElement) – The artifact to get the MIME type for.
Returns:: The MIME type of the artifact.
Return type:: str

aignostics.platform.mime_type_to_file_ending(mime_type: str) → str¶

Converts a MIME type to an appropriate file extension.

Parameters:: mime_type (str) – The MIME type string to convert.
Returns:: The corresponding file extension including the dot.
Return type:: str
Raises:: ValueError – If the MIME type is not recognized.

aignostics.platform.settings() → Settings¶

Lazy load authentication settings from the environment or a file.

Given we use Pydantic Settings, validation is done automatically.
We only load and validate if we actually need the settings,
thereby not killing the client on other actions.
If the settings have already been loaded, return the cached instance.

Returns:: The loaded authentication settings.
Return type:: AuthenticationSettings

Application module.

pydantic model aignostics.application.DownloadProgress¶

Model for tracking download progress with computed progress metrics.

Fields:

artifact (aignx.codegen.models.output_artifact_result_read_response.OutputArtifactResultReadResponse | None)
artifact_count (int | None)
artifact_download_url (str | None)
artifact_downloaded_chunk_size (int)
artifact_downloaded_size (int)
artifact_index (int | None)
artifact_path (pathlib.Path | None)
artifact_size (int | None)
input_slide_downloaded_chunk_size (int)
input_slide_downloaded_size (int)
input_slide_path (pathlib.Path | None)
input_slide_size (int | None)
input_slide_url (str | None)
item (aignx.codegen.models.item_result_read_response.ItemResultReadResponse | None)
item_count (int | None)
item_external_id (str | None)
item_index (int | None)
qupath_add_input_progress (aignostics.qupath._service.AddProgress | None)
qupath_add_results_progress (aignostics.qupath._service.AddProgress | None)
qupath_annotate_input_with_results_progress (aignostics.qupath._service.AnnotateProgress | None)
run (aignx.codegen.models.run_read_response.RunReadResponse | None)
status (aignostics.application._models.DownloadProgressState)

field artifact: OutputArtifactResultReadResponse | None = None¶

field artifact_count: int | None = None¶

field artifact_download_url: str | None = None¶

field artifact_downloaded_chunk_size: int = 0¶

field artifact_downloaded_size: int = 0¶

field artifact_index: int | None = None¶

field artifact_path: Path | None = None¶

field artifact_size: int | None = None¶

field input_slide_downloaded_chunk_size: int = 0¶

field input_slide_downloaded_size: int = 0¶

field input_slide_path: Path | None = None¶

field input_slide_size: int | None = None¶

field input_slide_url: str | None = None¶

field item: ItemResultReadResponse | None = None¶

field item_count: int | None = None¶

field item_external_id: str | None = None¶

field item_index: int | None = None¶

field qupath_add_input_progress: AddProgress | None = None¶

field qupath_add_results_progress: AddProgress | None = None¶

field qupath_annotate_input_with_results_progress: AnnotateProgress | None = None¶

field run: RunReadResponse | None = None¶

field status: DownloadProgressState = DownloadProgressState.INITIALIZING¶

property artifact_progress_normalized: float¶

Compute normalized artifact progress in range 0..1.

Returns:: The normalized artifact progress in range 0..1.
Return type:: float

property item_progress_normalized: float¶

Compute normalized item progress in range 0..1.

Returns:: The normalized item progress in range 0..1.
Return type:: float

property total_artifact_count: int | None¶

Calculate total number of artifacts across all items.

Returns:: Total artifact count or None if counts not available.
Return type:: int | None

property total_artifact_index: int | None¶

Calculate the current artifact index across all items.

Returns:: Total artifact index or None if indices not available.
Return type:: int | None

class aignostics.application.DownloadProgressState(*values)¶: Enum for download progress states.

class aignostics.application.PageBuilder¶

static register_pages() → None¶: Register pages.

class aignostics.application.Service¶

Service of the application module.

application(application_id: str) → ApplicationReadResponse¶

Get application.

Parameters:

application_id (str) – The ID of the application.

Returns:

The application.

Return type:

Application

Raises:

NotFoundException – If the application with the given ID is not found.
RuntimeError – If the application cannot be retrieved unexpectedly.

application_run(run_id: str) → Run¶

Select a run by its ID.

Parameters:: run_id (str) – The ID of the run to find
Returns:: The run that can be fetched using the .details() call.
Return type:: Run
Raises:: RuntimeError – If initializing the client fails or the run cannot be retrieved.

application_run_cancel(run_id: str) → None¶

Cancel a run by its ID.

Parameters:

run_id (str) – The ID of the run to cancel

Raises:

Exception – If the client cannot be created.
NotFoundException – If the application run with the given ID is not found.
ValueError – If the run ID is invalid or the run cannot be canceled given its current state.
RuntimeError – If canceling the run fails unexpectedly.

application_run_delete(run_id: str) → None¶

Delete a run by its ID.

Parameters:

run_id (str) – The ID of the run to delete

Raises:

Exception – If the client cannot be created.
NotFoundException – If the application run with the given ID is not found.
ValueError – If the run ID is invalid or the run cannot be deleted given its current state.
RuntimeError – If deleting the run fails unexpectedly.

application_run_download(run_id: str, destination_directory: Path, create_subdirectory_for_run: bool = True, create_subdirectory_per_item: bool = True, wait_for_completion: bool = True, qupath_project: bool = False, download_progress_queue: Any | None = None, download_progress_callable: Callable | None = None) → Path¶

Download application run results with progress tracking.

Parameters:

run_id (str) – The ID of the application run to download.
destination_directory (Path) – Directory to save downloaded files.
create_subdirectory_for_run (bool) – Whether to create a subdirectory for the run.
create_subdirectory_per_item (bool) – Whether to create a subdirectory for each item, if not set, all items will be downloaded to the same directory but prefixed with the item external id and underscore.
wait_for_completion (bool) – Whether to wait for run completion. Defaults to True.
qupath_project (bool) – If True, create QuPath project referencing input slides and results. This requires QuPath to be installed. The QuPath project will be created in a subfolder of the destination directory.
download_progress_queue (Queue | None) – Queue for GUI progress updates.
download_progress_callable (Callable | None) – Callback for CLI progress updates.

Returns:

The directory containing downloaded results.

Return type:

Path

Raises:

ValueError – If the run ID is invalid or destination directory cannot be created or QuPath extra is not installed when qupath_project=True.
NotFoundException – If the application run with the given ID is not found.
RuntimeError – If run details cannot be retrieved or download fails.

static application_run_download_static(run_id: str, destination_directory: Path, create_subdirectory_for_run: bool = True, create_subdirectory_per_item: bool = True, wait_for_completion: bool = True, qupath_project: bool = False, download_progress_queue: Any | None = None) → Path¶

Download application run results with progress tracking, static variant.

Parameters:

run_id (str) – The ID of the application run to download.
destination_directory (Path) – Directory to save downloaded files.
create_subdirectory_for_run (bool) – Whether to create a subdirectory for the run.
create_subdirectory_per_item (bool) – Whether to create a subdirectory for each item, if not set, all items will be downloaded to the same directory but prefixed with the item external ID and underscore.
wait_for_completion (bool) – Whether to wait for run completion. Defaults to True.
qupath_project (bool) – If True, create QuPath project referencing input slides and results. This requires QuPath to be installed. The QuPath project will be created in a subfolder of the destination directory.
download_progress_queue (Queue | None) – Queue for GUI progress updates.

Returns:

The directory containing downloaded results.

Return type:

Path

Raises:

ValueError – If the run ID is invalid or destination directory cannot be created or QuPath extra is not installed when qupath_project=True.
NotFoundException – If the application run with the given ID is not found.
RuntimeError – If run details cannot be retrieved or download fails.

application_run_submit(application_id: str, items: list[ItemCreationRequest], application_version: str | None = None, custom_metadata: dict[str, Any] | None = None, note: str | None = None, tags: set[str] | None = None, due_date: str | None = None, deadline: str | None = None, onboard_to_aignostics_portal: bool = False, gpu_type: str | None = None, gpu_provisioning_mode: str | None = None, max_gpus_per_slide: int | None = None, flex_start_max_run_duration_minutes: int | None = None, cpu_provisioning_mode: str | None = None, node_acquisition_timeout_minutes: int | None = None) → Run¶

Submit a run for the given application.

Parameters:

application_id (str) – The ID of the application to run.
items (list[InputItem]) – The input items for the run.
application_version (str | None) – The version of the application to run.
custom_metadata (dict[str, Any] | None) – Optional custom metadata to attach to the run.
note (str | None) – An optional note for the run.
tags (set[str] | None) – Optional set of tags to attach to the run for filtering.
due_date (str | None) – An optional requested completion time for the run, ISO8601 format. The scheduler will try to complete the run before this time, taking the subscription tier and available GPU resources into account.
deadline (str | None) – An optional hard deadline for the run, ISO8601 format. If processing exceeds this deadline, the run can be aborted.
onboard_to_aignostics_portal (bool) – True if the run should be onboarded to the Aignostics Portal.
gpu_type (str | None) – The type of GPU to use (L4 or A100).
gpu_provisioning_mode (str | None) – The provisioning mode for GPU resources (SPOT, ON_DEMAND, or FLEX_START).
max_gpus_per_slide (int | None) – The maximum number of GPUs to allocate per slide.
flex_start_max_run_duration_minutes (int | None) – Maximum run duration in minutes when using FLEX_START provisioning mode (1-3600).
cpu_provisioning_mode (str | None) – The provisioning mode for CPU resources (SPOT or ON_DEMAND).
node_acquisition_timeout_minutes (int | None) – Timeout for acquiring compute nodes in minutes.

Returns:

The submitted run.

Return type:

Run

Raises:

NotFoundException – If the application version with the given ID is not found.
ValueError – If the application version ID is invalid or items invalid or due_date not ISO 8601 or due_date not in the future or deadline not ISO 8601 or deadline not in the future or due_date not before deadline.
RuntimeError – If submitting the run failed unexpectedly.

application_run_submit_from_metadata(application_id: str, metadata: list[dict[str, Any]], application_version: str | None = None, custom_metadata: dict[str, Any] | None = None, note: str | None = None, tags: set[str] | None = None, due_date: str | None = None, deadline: str | None = None, onboard_to_aignostics_portal: bool = False, gpu_type: str | None = None, gpu_provisioning_mode: str | None = None, max_gpus_per_slide: int | None = None, flex_start_max_run_duration_minutes: int | None = None, cpu_provisioning_mode: str | None = None, node_acquisition_timeout_minutes: int | None = None) → Run¶

Submit a run for the given application.

Parameters:

application_id (str) – The ID of the application to run.
metadata (list[dict[str, Any]]) – The metadata for the run.
custom_metadata (dict[str, Any] | None) – Optional custom metadata to attach to the run.
note (str | None) – An optional note for the run.
tags (set[str] | None) – Optional set of tags to attach to the run for filtering.
due_date (str | None) – An optional requested completion time for the run, ISO8601 format. The scheduler will try to complete the run before this time, taking the subscription tier and available GPU resources into account.
deadline (str | None) – An optional hard deadline for the run, ISO8601 format. If processing exceeds this deadline, the run can be aborted.
application_version (str | None) – The version of the application. If not given latest version is used.
onboard_to_aignostics_portal (bool) – True if the run should be onboarded to the Aignostics Portal.
gpu_type (str | None) – The type of GPU to use (L4 or A100).
gpu_provisioning_mode (str | None) – The provisioning mode for GPU resources (SPOT, ON_DEMAND, or FLEX_START).
max_gpus_per_slide (int | None) – The maximum number of GPUs to allocate per slide.
flex_start_max_run_duration_minutes (int | None) – Maximum run duration in minutes when using FLEX_START provisioning mode (1-3600).
cpu_provisioning_mode (str | None) – The provisioning mode for CPU resources (SPOT or ON_DEMAND).
node_acquisition_timeout_minutes (int | None) – Timeout for acquiring compute nodes in minutes.

Returns:

The submitted run.

Return type:

Run

Raises:

NotFoundException – If the application version with the given ID is not found.
ValueError – If platform bucket URL is missing or has unsupported protocol, or if the application version ID is invalid, or if due_date is not ISO 8601 or if due_date not in the future.
RuntimeError – If submitting the run failed unexpectedly.

application_run_update_custom_metadata(run_id: str, custom_metadata: dict[str, Any]) → None¶

Update custom metadata for an existing application run.

Parameters:

run_id (str) – The ID of the run to update
custom_metadata (dict[str, Any]) – The new custom metadata to attach to the run.

Raises:

NotFoundException – If the application run with the given ID is not found.
ValueError – If the run ID is invalid.
RuntimeError – If updating the run metadata fails unexpectedly.

static application_run_update_custom_metadata_static(run_id: str, custom_metadata: dict[str, Any]) → None¶

Static wrapper for updating custom metadata for an application run.

Parameters:

run_id (str) – The ID of the run to update
custom_metadata (dict[str, Any]) – The new custom metadata to attach to the run.

Raises:

NotFoundException – If the application run with the given ID is not found.
ValueError – If the run ID is invalid.
RuntimeError – If updating the run metadata fails unexpectedly.

application_run_update_item_custom_metadata(run_id: str, external_id: str, custom_metadata: dict[str, Any]) → None¶

Update custom metadata for an existing item in an application run.

Parameters:

run_id (str) – The ID of the run containing the item
external_id (str) – The external ID of the item to update
custom_metadata (dict[str, Any]) – The new custom metadata to attach to the item.

Raises:

NotFoundException – If the application run or item with the given IDs is not found.
ValueError – If the run ID or item external ID is invalid.
RuntimeError – If updating the item metadata fails unexpectedly.

static application_run_update_item_custom_metadata_static(run_id: str, external_id: str, custom_metadata: dict[str, Any]) → None¶

Static wrapper for updating custom metadata for an item in an application run.

Parameters:

run_id (str) – The ID of the run containing the item
external_id (str) – The external ID of the item to update
custom_metadata (dict[str, Any]) – The new custom metadata to attach to the item.

Raises:

NotFoundException – If the application run or item with the given IDs is not found.
ValueError – If the run ID or item external ID is invalid.
RuntimeError – If updating the item metadata fails unexpectedly.

static application_run_upload(application_id: str, metadata: list[dict[str, Any]], application_version: str | None = None, onboard_to_aignostics_portal: bool = False, upload_prefix: str = '1775719937415.5989', upload_progress_queue: Any | None = None, upload_progress_callable: Callable[[int, Path, str], None] | None = None) → bool¶

Upload files with a progress queue.

Parameters:

application_id (str) – The ID of the application.
metadata (list[dict[str, Any]]) – The metadata to upload.
application_version (str|None) – The version ID of the application. If not given latest version is used.
onboard_to_aignostics_portal (bool) – True if the run should be onboarded to the Aignostics Portal.
upload_prefix (str) – The prefix for the upload, defaults to current milliseconds.
upload_progress_queue (Queue | None) – The queue to send progress updates to.
upload_progress_callable (Callable[[int, Path, str], None] | None) – The task to update for progress updates.

Returns:

True if the upload was successful, False otherwise.

Return type:

bool

Raises:

NotFoundException – If the application version with the given ID is not found.
RuntimeError – If fetching the application version fails unexpectedly.
requests.HTTPError – If the upload fails with an HTTP error.

Get a list of all application runs.

Parameters:

application_id (str | None) – The ID of the application to filter runs. If None, no filtering is applied.
application_version (str | None) – The version of the application to filter runs. If None, no filtering is applied.
external_id (str | None) – The external ID to filter runs. If None, no filtering is applied.
has_output (bool) – If True, only runs with partial or full output are retrieved.
note_regex (str | None) – Optional regex to filter runs by note metadata. If None, no filtering is applied. Cannot be used together with query parameter.
note_query_case_insensitive (bool) – If True, the note_regex is case insensitive. Default is True.
tags (set[str] | None) – Optional set of tags to filter runs. All tags must match. If None, no filtering is applied. Cannot be used together with query parameter.
query (str | None) – Optional string to filter runs by note OR tags (case insensitive partial match). If None, no filtering is applied. Cannot be used together with custom_metadata, note_regex, or tags. Performs a union search: matches runs where the query appears in the note OR matches any tag.
limit (int | None) – The maximum number of runs to retrieve. If None, all runs are retrieved.
for_organization (str | None) – If set, returns all runs triggered by users of the specified organization. If None, only the runs of the current user are returned.

Returns:

A list of all application runs.

Return type:

list[RunData]

Raises:

ValueError – If query is used together with custom_metadata, note_regex, or tags.
ForbiddenException – If the user is not authorized to list runs for the specified organization.
RuntimeError – If the application run list cannot be retrieved.

static application_runs_static(application_id: str | None = None, application_version: str | None = None, external_id: str | None = None, has_output: bool = False, note_regex: str | None = None, note_query_case_insensitive: bool = True, tags: set[str] | None = None, query: str | None = None, limit: int | None = None, for_organization: str | None = None) → list[dict[str, Any]]¶

Get a list of all application runs, static variant.

Parameters:

application_id (str | None) – The ID of the application to filter runs. If None, no filtering is applied.
application_version (str | None) – The version of the application to filter runs. If None, no filtering is applied.
external_id (str | None) – The external ID to filter runs. If None, no filtering is applied.
has_output (bool) – If True, only runs with partial or full output are retrieved.
note_regex (str | None) – Optional regex to filter runs by note metadata. If None, no filtering is applied. Cannot be used together with query parameter.
note_query_case_insensitive (bool) – If True, the note_regex is case insensitive. Default is True.
tags (set[str] | None) – Optional set of tags to filter runs. All tags must match. Cannot be used together with query parameter.
query (str | None) – Optional string to filter runs by note OR tags (case insensitive partial match). If None, no filtering is applied. Cannot be used together with custom_metadata, note_regex, or tags. Performs a union search: matches runs where the query appears in the note OR matches any tag.
limit (int | None) – The maximum number of runs to retrieve. If None, all runs are retrieved.
for_organization (str | None) – If set, returns all runs triggered by users of the specified organization. If None, only the runs of the current user are returned.

Returns:

A list of all application runs.

Return type:

list[RunData]

Raises:

ValueError – If query is used together with custom_metadata, note_regex, or tags.
RuntimeError – If the application run list cannot be retrieved.

application_version(application_id: str, application_version: str | None = None) → VersionReadResponse¶

Get a specific application version.

Parameters:

application_id (str) – The ID of the application
application_version (str|None) – The version of the application (semver). If not given latest version is used.

Returns:

The application version

Return type:

ApplicationVersion

Raises:

ValueError – If the application version number is invalid.
NotFoundException – If the application version with the given ID and number is not found.
RuntimeError – If the application cannot be retrieved unexpectedly.

application_versions(application_id: str) → list[VersionReadResponse]¶

Get a list of all versions for a specific application.

Parameters:

application_id (str) – The ID of the application.

Returns:

A list of all versions for the application.

Return type:

list[ApplicationVersion]

Raises:

RuntimeError – If the versions cannot be retrieved unexpectedly.
NotFoundException – If the application with the given ID is not found.

static application_versions_static(application_id: str) → list[VersionReadResponse]¶

Get a list of all versions for a specific application, static variant.

Parameters:: application_id (str) – The ID of the application.
Returns:: A list of all versions for the application.
Return type:: list[ApplicationVersion]
Raises:: Exception – If the application versions cannot be retrieved.

applications() → list[ApplicationReadShortResponse]¶

Get a list of all applications.

Returns:

A list of all applications.

Return type:

list[str]

Raises:

Exception – If the client cannot be created.
Exception – If the application list cannot be retrieved.

static applications_static() → list[ApplicationReadShortResponse]¶

Get a list of all applications, static variant.

Returns:

A list of all applications.

Return type:

list[str]

Raises:

Exception – If the client cannot be created.
Exception – If the application list cannot be retrieved.

static generate_metadata_from_source_directory(source_directory: Path, application_id: str, application_version: str | None = None, with_gui_metadata: bool = False, mappings: list[str] | None = None, with_extra_metadata: bool = False) → list[dict[str, Any]]¶

Generate metadata from the source directory.

Steps: 1. Recursively scans files ending with supported extensions in the source directory 2. For DICOM files (.dcm), filters out auxiliary and redundant files 3. Creates a dict for each file with the following fields:

external_id (str): The external_id of the file, by default equivalent to the absolute file name

source (str): The absolute filename

checksum_base64_crc32c (str): The CRC32C checksum of the file, base64 encoded

resolution_mpp (float): The microns per pixel, inspecting the base layer

height_px (int): The height of the image in pixels, inspecting the base layer

width_px (int): The width of the image in pixels, inspecting the base layer

Further attributes depending on the application and its version

Applies the optional mappings to fill in additional metadata fields in the dict

Parameters:

source_directory – The source directory to generate metadata from.
application_id – The ID of the application.
application_version – The version of the application (semver). If not given, latest version is used.
with_gui_metadata – If True, include additional metadata for GUI display.
mappings – Mappings of the form ‘<regexp>:<key>=<value>,<key>=<value>,…’. The regular expression is matched against the external_id attribute of the entry. The key/value pairs are applied to the entry if the pattern matches.
with_extra_metadata – If True, include extra metadata from the WSIService.

Returns:

List of metadata dictionaries, one per processable file found.

Raises:

NotFoundException – If the application version with the given ID is not found.
ValueError – If the source directory does not exist or is not a directory.
RuntimeError – If the metadata generation fails unexpectedly.

async health() → Health¶

Determine health of this service.

Returns:: The health of the service.
Return type:: Health

async info(mask_secrets: bool = True) → dict[str, Any]¶

Determine info of this service.

Parameters:: mask_secrets (bool) – If True, mask sensitive information in the output.
Returns:: The info of this service.
Return type:: dict[str,Any]

pydantic settings aignostics.application.Settings¶

Settings.

Show JSON schema

{
   "title": "Settings",
   "description": "Settings.",
   "type": "object",
   "properties": {}
}

Config:

extra: str = ignore
env_prefix: str = AIGNOSTICS_APPLICATION_
env_file: list = [PosixPath(‘/home/docs/.aignostics/.env’), PosixPath(‘/home/docs/.aignostics/.env.local’), PosixPath(‘.env’), PosixPath(‘.env.local’)]
env_file_encoding: str = utf-8

Dataset module.

class aignostics.dataset.PageBuilder¶

static register_pages() → None¶: Register pages.

class aignostics.dataset.Service(settings_class: type[T] | None = None)¶

Service of the IDC module.

static download_aignostics(source_url: str, destination_directory: Path, download_progress_callable: Callable[[int, int, str], None] | None = None) → Path¶

Download from bucket to folder via a bucket URL.

Parameters:

source_url (str) – URL to download, e.g. gs://aignostics-platform-ext-a4f7e9/python-sdk-tests/he-tme/slides/…
destination_directory (Path) – Destination directory to download to.
download_progress_callable (Callable[[int, int, str], None] | None) – Optional callback for progress updates. Called with (bytes_downloaded, total_size, filename).

Returns:

The path to the downloaded file.

Return type:

Path

Raises:

ValueError – If the source URL is invalid.
RuntimeError – If the download fails.

static download_idc(source: str, target: Path, target_layout: str = '%collection_id/%PatientID/%StudyInstanceUID/%Modality_%SeriesInstanceUID/', dry_run: bool = False) → int¶

Download from IDC using identifier or comma-separated set of identifiers.

Parameters:

source (str) – Identifier or comma-separated set of identifiers. IDs matched against collection_id, PatientId, StudyInstanceUID, SeriesInstanceUID or SOPInstanceUID.
target (Path) – Target directory for download.
target_layout (str) – Layout of the target directory.
dry_run (bool) – If True, perform a dry run.

Returns:

Number of matched identifier types.

Return type:

int

Raises:

ValueError – If target directory does not exist or no valid IDs provided.
RuntimeError – If download fails.

static download_with_queue(queue: Queue, source: str, target: str = '/home/docs/checkouts/readthedocs.org/user_builds/aignostics/checkouts/latest/docs/source', target_layout: str = '%collection_id/%PatientID/%StudyInstanceUID/%Modality_%SeriesInstanceUID/', dry_run: bool = False) → None¶

Download from manifest file, identifier, or comma-separate set of identifiers.

Parameters:

queue (Queue) – The queue to use for progress updates.
source (str) – The source to download from.
target (str) – The target directory to download to.
target_layout (str) – The layout of the target directory.
dry_run (bool) – If True, perform a dry run.

Raises:

ValueError – If the target directory does not exist.

async health() → Health¶

Determine health of hello service.

Returns:: The health of the service.
Return type:: Health

async info(mask_secrets: bool = True) → dict[str, Any]¶

Determine info of this service.

Parameters:: mask_secrets (bool) – Whether to mask sensitive information in the output.
Returns:: The info of this service.
Return type:: dict[str,Any]

Bucket module.

class aignostics.bucket.PageBuilder¶

static register_pages() → None¶: Register pages.

class aignostics.bucket.Service¶

Service of the bucket module.

create_signed_download_url(object_key: str, bucket_name: str | None = None) → str¶

Generates a signed URL to download a Google Cloud Storage object.

Parameters:

object_key (str) – The key of the object to generate a signed URL for.
bucket_name (str | None) – The name of the bucket to generate a signed URL for. If None, use the default bucket.

Returns:

A signed URL that can be used to download from the bucket and key.

Return type:

str

create_signed_upload_url(object_key: str, bucket_name: str | None = None) → str¶

Generates a signed URL to upload a Google Cloud Storage object.

Parameters:

object_key (str) – The key of the object to generate a signed URL for.
bucket_name (str) – The name of the bucket to generate a signed URL for. If None, use the default bucket.

Returns:

A signed URL that can be used to upload to the bucket and key.

Return type:

str

delete(what: list[str] | None, what_is_key: bool = False, dry_run: bool = True) → int¶

Delete objects.

Parameters:

what (list[str] | None) – Patterns or keys to match object keys against - all if not specified.
what_is_key (bool) – If True, treat the entries as exact keys, else as regex patterns.
dry_run (bool) – If True, perform a dry run without actual deletion.

Returns:

Number of objects deleted.

Return type:

int

Raises:

ValueError – If any provided regex pattern is invalid.

static delete_static(what: list[str] | None, what_is_key: bool = False, dry_run: bool = True) → int¶

Delete objects, static variant.

Parameters:

what (list[str] | None) – Patterns or keys to match object keys against - all if not specified.
what_is_key (bool) – If True, treat the entries as exact keys, else as regex patterns.
dry_run (bool) – If True, perform a dry run without actual deletion.

Returns:

Number of objects deleted.

Return type:

int

Raises:

ValueError – If any provided regex pattern is invalid.

download(what: list[str] | None = None, destination: Path = PosixPath('/home/docs/.local/share/aignostics/bucket_downloads'), what_is_key: bool = False, progress_callback: Callable[[DownloadProgress], None] | None = None) → DownloadResult¶

Download files matching a pattern from the bucket.

Parameters:

what (list[str] | None) – Patterns or keys to match object keys against - all if not specified.
what_is_key (bool) – If True, treat the pattern as a key, else as a regex.
destination (Path) – Destination directory for downloaded files.
progress_callback (Callable[[DownloadProgress], None] | None) – Optional callback for download progress.

Returns:

Result containing lists of successfully downloaded and failed objects.

Return type:

DownloadResult

Raises:

ValueError – If any provided regex pattern is invalid.

static download_static(what: list[str] | None = None, destination: Path = PosixPath('/home/docs/.local/share/aignostics/bucket_downloads'), what_is_key: bool = False, progress_callback: Callable[[DownloadProgress], None] | None = None) → DownloadResult¶

Download files matching a pattern from the bucket, static method version.

Parameters:

what (list[str] | None) – Patterns or keys to match object keys against - all if not specified.
what_is_key (bool) – If True, treat the pattern as a key, else as a regex.
destination (Path) – Destination directory for downloaded files.
progress_callback (Callable[[DownloadProgress], None] | None) – Optional callback for download progress.

Returns:

Result containing lists of successfully downloaded and failed objects.

Return type:

DownloadResult

Raises:

ValueError – If any provided regex pattern is invalid.

find(what: list[str] | None, what_is_key: bool = False, detail: bool = False, include_signed_urls: bool = False) → list[str | dict[str, Any]]¶

List objects recursively in the bucket, with optional pattern matching.

Parameters:

what (list[str] | None) – Patterns or keys to match object keys against - all if not specified.
what_is_key (bool) – If True, treat entries in what as exact keys, else as regex patterns.
detail (bool) – If True, return detailed information including object type, else return only paths.
include_signed_urls (bool) – If True, include signed download URLs in the results.

Returns:

List of objects in the bucket with optional detail.

Return type:

list[Union[str, dict[str, Any]]]

Raises:

ValueError – If any provided regex pattern is invalid.

static find_static(what: list[str] | None = None, what_is_key: bool = False, detail: bool = False, include_signed_urls: bool = False) → list[str | dict[str, Any]]¶

List objects recursively in the bucket, static method version.

Parameters:

what (list[str] | None) – Patterns or keys to match object keys against - all if not specified.
what_is_key (bool) – If True, treat the pattern as a key, else as a regex.
detail (bool) – If True, return detailed information including object type, else return only paths.
include_signed_urls (bool) – If True, include signed download URLs in the results.

Returns:

List of objects in the bucket with optional detail.

Return type:

list[Union[str, dict[str, Any]]]

Raises:

ValueError – If the provided regex pattern is invalid.

get_bucket_name() → str¶

Get the bucket name.

Returns:: The bucket name.
Return type:: str
Raises:: ValueError – If the bucket name is not set in the organization settings.

static get_bucket_protocol() → str¶

Get the bucket protocol.

Returns:: The bucket protocol.
Return type:: str

async health() → Health¶

Determine health of this service.

Returns:: The health of the service.
Return type:: Health

async info(mask_secrets: bool = True) → dict[str, Any]¶

Determine info of this service.

Parameters:: mask_secrets (bool) – If True, mask sensitive information in the output.
Returns:: The info of this service.
Return type:: dict[str,Any]

upload(source_path: Path, destination_prefix: str, callback: Callable[[int, Path], None] | None = None) → dict[str, list[str]]¶

Upload a file or directory to the bucket.

Parameters:

source_path (Path) – Path to file or directory to upload.
destination_prefix (str) – Prefix for object keys (e.g. username).
callback (Callable[[int, int], None] | None) – Optional callback function for upload progress. Function receives bytes_read and total_bytes parameters.

Returns:

Dict with ‘success’ and ‘failed’ lists containing object keys.

Return type:

dict[str, list[str]]

pydantic settings aignostics.bucket.Settings¶

Settings.

Show JSON schema

{
   "title": "Settings",
   "description": "Settings.",
   "type": "object",
   "properties": {
      "protocol": {
         "$ref": "#/$defs/BucketProtocol",
         "default": "gs",
         "description": "Protocol to access the cloud bucket, default is gs"
      },
      "region_name": {
         "default": "EUROPE-WEST3",
         "description": "Region of the cloud bucket",
         "title": "Region Name",
         "type": "string"
      },
      "upload_signed_url_expiration_seconds": {
         "default": 7200,
         "description": "Expiration time for signed URLs created and used by the Python SDK to upload to the bucket.",
         "maximum": 604800,
         "minimum": 60,
         "title": "Upload Signed Url Expiration Seconds",
         "type": "integer"
      },
      "download_signed_url_expiration_seconds": {
         "default": 604800,
         "description": "Expiration time of the signed URLs provided by the Python SDK to the platform on submitting application runs.",
         "maximum": 604800,
         "minimum": 60,
         "title": "Download Signed Url Expiration Seconds",
         "type": "integer"
      }
   },
   "$defs": {
      "BucketProtocol": {
         "enum": [
            "gs",
            "s3"
         ],
         "title": "BucketProtocol",
         "type": "string"
      }
   }
}

Config:

extra: str = ignore
env_prefix: str = AIGNOSTICS_BUCKET_
env_file: list = [PosixPath(‘/home/docs/.aignostics/.env’), PosixPath(‘/home/docs/.aignostics/.env.local’), PosixPath(‘.env’), PosixPath(‘.env.local’)]
env_file_encoding: str = utf-8

Fields:

download_signed_url_expiration_seconds (int)
protocol (aignostics.bucket._settings.BucketProtocol)
region_name (str)
upload_signed_url_expiration_seconds (int)

field download_signed_url_expiration_seconds: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=604800, description='Expiration time of the signed URLs provided by the Python SDK to the platform on submitting application runs.', metadata=[Ge(ge=60), Le(le=604800)])] = 604800¶

Expiration time of the signed URLs provided by the Python SDK to the platform on submitting application runs.

Constraints:

ge = 60
le = 604800

field protocol: Annotated[BucketProtocol, FieldInfo(annotation=NoneType, required=False, default='gs', description='Protocol to access the cloud bucket, default is gs')] = 'gs'¶: Protocol to access the cloud bucket, default is gs

field region_name: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='EUROPE-WEST3', description='Region of the cloud bucket')] = 'EUROPE-WEST3'¶: Region of the cloud bucket

field upload_signed_url_expiration_seconds: Annotated[int, FieldInfo(annotation=NoneType, required=False, default=7200, description='Expiration time for signed URLs created and used by the Python SDK to upload to the bucket.', metadata=[Ge(ge=60), Le(le=604800)])] = 7200¶

Expiration time for signed URLs created and used by the Python SDK to upload to the bucket.

Constraints:

ge = 60
le = 604800

Notebook module.

class aignostics.notebook.PageBuilder¶

static register_pages() → None¶: Register pages.

class aignostics.notebook.Service(settings_class: type[T] | None = None)¶

Service of the Marimo module.

async health() → Health¶

Determine health of hello service.

Returns:: The health of the service.
Return type:: Health

async info(mask_secrets: bool = True) → dict[str, Any]¶

Determine info of this service.

Parameters:: mask_secrets (bool) – Whether to mask sensitive information in the output.
Returns:: The info of this service.
Return type:: dict[str,Any]

is_marimo_server_running() → bool¶

Check if the marimo server is running.

Returns:: True if the server is running, False otherwise.
Return type:: bool

is_monitor_thread_alive() → bool¶

Check if the monitor thread is running.

Returns:: True if the thread is running, False otherwise.
Return type:: bool

start() → str¶

Start the Marimo server.

Returns:: The URL of the started Marimo server.
Return type:: str
Raises:: RuntimeError – If the Marimo server fails to start or if the URL isn’t detected within 10 seconds.

stop() → None¶: Stop the Marimo server.

WSI module.

class aignostics.wsi.PageBuilder¶

static register_pages() → None¶: Register pages.

class aignostics.wsi.Service(settings_class: type[T] | None = None)¶

Service of the application module.

static get_metadata(path: Path) → dict[str, Any]¶

Get metadata from a TIFF file.

Parameters:

path (Path) – Path to the TIFF file.

Returns:

Metadata of the TIFF file.

Return type:

dict[str, Any]

Raises:

ValueError – If the file type is not supported.
RuntimeError – If there is an error processing the file.

static get_thumbnail(path: Path, max_safe_dimension: int = 4096) → PIL.Image.Image¶

Get thumbnail as PIL image.

Parameters:

path (Path) – Path to the image.
max_safe_dimension (int) – Maximum dimension (width or height) of smallest pyramid level before considering the pyramid incomplete.

Returns:

Thumbnail of the image.

Return type:

PIL.Image.Image

Raises:

ValueError – If the file type is not supported.
RuntimeError – If there is an error generating the thumbnail.

static get_thumbnail_bytes(path: Path, max_safe_dimension: int = 4096) → bytes¶

Get thumbnail of a image as bytes.

Parameters:

path (Path) – Path to the image.
max_safe_dimension (int) – Maximum dimension (width or height) of smallest pyramid level before considering the pyramid incomplete.

Returns:

Thumbnail of the image.

Return type:

bytes

Raises:

ValueError – If the file type is not supported.
RuntimeError – If there is an error processing the file.

static get_tiff_as_jpg(url: str) → bytes¶

Get a TIFF image from a URL and convert it to JPG format.

Parameters:

url (str) – URL to the TIFF image.

Returns:

The TIFF image converted to JPG format as bytes.

Return type:

bytes

Raises:

ValueError – If URL format is invalid or if there’s an error opening the tiff.
RuntimeError – If there’s an unexpected internal error.

static get_wsi_files_to_process(path: Path, extension: str) → Iterable[Path]¶

Get WSI files to process for the specified extension.

For DICOM files (.dcm), applies filtering to only include WSI files and select only the highest resolution file from multi-file pyramids. For other formats, returns all files matching the extension.

Parameters:

path – Root directory to search for WSI files.
extension – File extension to filter (e.g., “.dcm”, “.tiff”, “.svs”). Must include the leading dot.

Returns:

Iterable of Path objects for files to process.

async health() → Health¶

Determine health of thumbnail service.

Returns:: The health of the service.
Return type:: Health

async info(mask_secrets: bool = True) → dict[str, Any]¶

Determine info of this service.

Parameters:: mask_secrets (bool) – Whether to mask sensitive information in the output.
Returns:: The info of this service.
Return type:: dict[str,Any]

System module.

class aignostics.system.PageBuilder¶

static register_pages() → None¶: Register pages.

class aignostics.system.Service¶

System service.

static dotenv_get(key: str) → str | None¶

Get value of key in environment.

Parameters:: key (str) – The key to add.
Returns:: The value of the key if it exists, None otherwise.
Return type:: str | None

static dotenv_set(key: str, value: str) → None¶

Set key-value pair in primary .env file, unset in alternative .env files.

Parameters:

key (str) – The key to add.
value (str) – The value to add.

Raises:

ValueError – If the primary .env file does not exist.

static dotenv_unset(key: str) → int¶

Unset key-value pair in all .env files.

Parameters:: key (str) – The key to remove.
Returns:: The number of times the key has been removed across files.
Return type:: int

static dump_dot_env_file(destination: Path) → None¶

Dump settings to .env file.

Parameters:: destination (Path) – Path pointing to .env file to generate.
Raises:: ValueError – If the primary .env file does not exist.

async health() → Health¶

Determine aggregate health of the system.

Health exposed by implementations of BaseService in other
modules is automatically included into the health tree.
See utils/_health.py:Health for an explanation of the health tree.

Returns:: The aggregate health of the system.
Return type:: Health

async static health_static() → Health¶

Determine health of the system.

This method is static and does not require an instance of the service.
It is used to determine the health of the system without needing to pass the service.

Returns:: The health of the system.
Return type:: Health

static http_proxy_disable() → None¶: Disable HTTP proxy.

static http_proxy_enable(host: str, port: int, scheme: str, ssl_cert_file: str | None = None, no_ssl_verify: bool = False) → None¶

Enable HTTP proxy.

Parameters:

host (str) – The host of the proxy server.
port (int) – The port of the proxy server.
scheme (str) – The scheme of the proxy server (e.g., “http”, “https”).
ssl_cert_file (str | None) – Path to the SSL certificate file, if any.
no_ssl_verify (bool) – Whether to disable SSL verification

Raises:

ValueError – If both ‘ssl_cert_file’ and ‘ssl_disable_verify’ are set.

async static info(include_environ: bool = False, mask_secrets: bool = True) → dict[str, Any]¶

Get info about configuration of service.

Runtime information is automatically compiled.
Settings are automatically aggregated from all implementations of
Pydantic BaseSettings in this package.
Info exposed by implementations of BaseService in other modules is
automatically included into the info dict.

Parameters:

include_environ (bool) – Whether to include environment variables in the info.
mask_secrets (bool) – Whether to mask information in environment variables identified as secrets

Returns:

Service configuration.

Return type:

dict[str, Any]

is_token_valid(token: str) → bool¶

Check if the presented token is valid.

Returns:: True if the token is valid, False otherwise.
Return type:: bool

Get OpenAPI schema of the webservice API provided by the platform.

Returns:: OpenAPI schema.
Return type:: dict[str, object]
Raises:: OpenAPISchemaError – If the OpenAPI schema file cannot be found or is not valid JSON.

static remote_diagnostics_disable() → None¶: Disable remote diagnostics.

static remote_diagnostics_enable() → None¶

Enable remote diagnostics via Sentry and Logfire. Data stored in EU data centers.

Raises:: ValueError – If the environment variable cannot be set.

static remote_diagnostics_enabled() → bool¶

Check if remote diagnostics are enabled.

Returns:: True if remote diagnostics are enabled, False otherwise.
Return type:: bool

pydantic settings aignostics.system.Settings¶

Settings.

Show JSON schema

{
   "title": "Settings",
   "description": "Settings.",
   "type": "object",
   "properties": {
      "token": {
         "anyOf": [
            {
               "format": "password",
               "type": "string",
               "writeOnly": true
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Secret token to present when performing sensitive operations such as retrieving info via webservice API",
         "title": "Token"
      }
   }
}

Config:

extra: str = ignore
env_prefix: str = AIGNOSTICS_SYSTEM_
env_file: list = [PosixPath(‘/home/docs/.aignostics/.env’), PosixPath(‘/home/docs/.aignostics/.env.local’), PosixPath(‘.env’), PosixPath(‘.env.local’)]
env_file_encoding: str = utf-8

Fields:

token (pydantic.types.SecretStr | None)

field token: Annotated[SecretStr | None, PlainSerializer(func=serialize_sensitive_info, return_type=str, when_used=always), FieldInfo(annotation=NoneType, required=False, default=None, description='Secret token to present when performing sensitive operations such as retrieving info via webservice API')] = None¶

Secret token to present when performing sensitive operations such as retrieving info via webservice API

Constraints:

func = <function OpaqueSettings.serialize_sensitive_info at 0x7de35fd9f320>
return_type = <class ‘str’>
when_used = always

Utilities module.

class aignostics.utils.BaseNavBuilder¶

Base class for navigation builders.

Each module should have ONE NavBuilder that defines its navigation items. NavBuilders are auto-discovered and used to populate the sidebar.

Example

class NavBuilder(BaseNavBuilder):

@staticmethod def get_nav_name() -> str:

return “My Module”

@staticmethod def get_nav_items() -> list[NavItem]:

return [
NavItem(icon=”home”, label=”Home”, target=”/my-module”), NavItem(icon=”settings”, label=”Settings”, target=”/my-module/settings”),

]

@staticmethod def get_nav_position() -> int:

return 200 # Lower = higher in sidebar

static get_nav_icon() → str¶

Return the icon for the navigation group expansion panel.

Uses Material Icons names. See: https://fonts.google.com/icons

Returns:: Material icon name. Defaults to ‘folder’.
Return type:: str

abstractmethod static get_nav_items() → list[NavItem]¶

Return navigation items for the sidebar.

Items are rendered in list order within the module’s group.

Returns:: Navigation items for this module.
Return type:: list[NavItem]

abstractmethod static get_nav_name() → str¶

Return the display name for this module’s navigation group.

Returns:: Display name shown in sidebar (e.g., ‘Hello World’, ‘System’).
Return type:: str

static get_nav_position() → int¶

Return position in sidebar (lower = higher).

Convention:

100-199: Core demo/example pages
200-499: Feature modules
500-799: Default (unspecified)
800-899: System/admin pages
900-999: External links (API docs, status, repo)

Returns:: Position value. Defaults to 1000.
Return type:: int

static get_nav_use_expansion() → bool¶

Return whether to render items in an expansion panel.

If True, items are grouped in a collapsible expansion with get_nav_name(). If False, items are rendered flat without grouping.

Returns:: Use expansion panel. Defaults to True.
Return type:: bool

class aignostics.utils.BasePageBuilder¶

Base class for all page builders.

abstractmethod static register_pages() → None¶: Register pages.

class aignostics.utils.BaseService(settings_class: type[T] | None = None)¶

Base class for services.

classmethod get_service() → Callable[[], Generator[Self]]¶

Create a FastAPI dependency that yields an instance of this service.

This is a factory method that returns a cached dependency function suitable for use with FastAPI’s Depends(). It eliminates boilerplate code by providing a standard pattern for service injection.

The dependency function is cached per-class to ensure the same function object is returned each time, which is necessary for FastAPI’s dependency_overrides to work correctly in tests.

Returns:: A dependency function that yields the service instance.
Return type:: Callable

Example

```python from aignostics.my_module._service import Service

@router.get(“/endpoint”) async def endpoint(service: Annotated[Service, Depends(Service.get_service())]):

return await service.do_something()

```

abstractmethod async health() → Health¶

Get health of this service. Override in subclass.

Returns:: Health status of the service.
Return type:: Health

abstractmethod async info(mask_secrets: bool = True) → dict[str, Any]¶

Get info of this service. Override in subclass.

Parameters:: mask_secrets – Whether to mask sensitive information in the output.
Returns:: Information about the service.
Return type:: dict[str, Any]

key() → str¶: Return the module name of the instance.

settings() → BaseSettings¶

Get the settings of this service.

Returns:: The settings of the service.
Return type:: BaseSettings

class aignostics.utils.GUILocalFilePicker(directory: str, *, upper_limit: str | EllipsisType | None = Ellipsis, multiple: bool = False, show_hidden_files: bool = False)¶

Local File Picker dialog class that lazy-loads NiceGUI dependencies.

static __new__(cls, directory: str, *, upper_limit: str | EllipsisType | None = Ellipsis, multiple: bool = False, show_hidden_files: bool = False) → GUILocalFilePicker¶

Create a new instance with lazy-loaded dependencies.

Parameters:

directory – The directory to start in.
upper_limit – The directory to stop at. None for no limit, default is same as starting directory.
multiple – Whether to allow multiple files to be selected.
show_hidden_files – Whether to show hidden files.

Returns:

An instance of the dialog with lazy-loaded dependencies.

pydantic model aignostics.utils.Health¶

Represents the health status of a service with optional components and failure reasons.

A health object can have child components, i.e. health forms a tree.
- Any node in the tree can set itself to DOWN or DEGRADED. If DOWN or DEGRADED, the node
  is required to set the reason attribute. If reason is not set when DOWN or DEGRADED, automatic model validation fails.
- DOWN trumps DEGRADED, DEGRADED trumps UP. If any child is DOWN, parent is DOWN.
  If none are DOWN but any are DEGRADED, parent is DEGRADED.
- The root of the health tree is computed in the system module.
  The health of other modules is automatically picked up by the system module.

Fields:

components (dict[str, aignostics.utils._health.Health])
reason (str | None)
status (aignostics.utils._health.HealthStatus)
uptime_statistics (dict[str, dict[str, Any]] | None)

Validators:

validate_health_state » all fields

field components: dict[str, Health] [Optional]¶

Validated by:

validate_health_state

field reason: str | None = None¶

Validated by:

validate_health_state

field status: HealthStatus [Required]¶

Validated by:

validate_health_state

field uptime_statistics: dict[str, dict[str, Any]] | None = None¶

Validated by:

validate_health_state

Code¶: alias of HealthStatus

compute_health_from_components() → Self¶

Recursively compute health status from components.

If health is already DOWN, it remains DOWN with its original reason.
If health is UP but any component is DOWN, health becomes DOWN with
a reason listing all failed components.
If no components are DOWN but any are DEGRADED, health becomes DEGRADED with a reason.

Returns:: The updated health instance with computed status.
Return type:: Self

validator validate_health_state » all fields¶

Validate the health state and ensure consistency.

Compute overall health based on component health
Ensure UP status has no associated reason
Ensure DOWN and DEGRADED status always have a reason

Returns:: The validated model instance with correct health status.
Return type:: Self
Raises:: ValueError – If validation fails due to inconsistency.

class aignostics.utils.HealthStatus(*values)¶

Health status enumeration for service health checks.

Values:: UP: Service is operating normally DEGRADED: Service is operational but with reduced functionality DOWN: Service is not operational

pydantic settings aignostics.utils.LogSettings¶

Settings for configuring logging behavior.

Show JSON schema

{
   "title": "LogSettings",
   "description": "Settings for configuring logging behavior.",
   "type": "object",
   "properties": {
      "level": {
         "default": "INFO",
         "description": "Log level, see https://loguru.readthedocs.io/en/stable/api/logger.html",
         "enum": [
            "CRITICAL",
            "ERROR",
            "WARNING",
            "SUCCESS",
            "INFO",
            "DEBUG",
            "TRACE"
         ],
         "title": "Level",
         "type": "string"
      },
      "stderr_enabled": {
         "default": true,
         "description": "Enable logging to stderr",
         "title": "Stderr Enabled",
         "type": "boolean"
      },
      "file_enabled": {
         "default": false,
         "description": "Enable logging to file",
         "title": "File Enabled",
         "type": "boolean"
      },
      "file_name": {
         "default": "/home/docs/.local/share/aignostics/aignostics.log",
         "description": "Name of the log file",
         "title": "File Name",
         "type": "string"
      },
      "redirect_logging": {
         "default": false,
         "description": "Redirect standard logging",
         "title": "Redirect Logging",
         "type": "boolean"
      }
   }
}

Config:

extra: str = ignore
env_prefix: str = AIGNOSTICS_LOG_
env_file: list = [PosixPath(‘/home/docs/.aignostics/.env’), PosixPath(‘/home/docs/.aignostics/.env.local’), PosixPath(‘.env’), PosixPath(‘.env.local’)]
env_file_encoding: str = utf-8

Fields:

file_enabled (bool)
file_name (str)
level (Literal['CRITICAL', 'ERROR', 'WARNING', 'SUCCESS', 'INFO', 'DEBUG', 'TRACE'])
redirect_logging (bool)
stderr_enabled (bool)

Validators:

validate_file_name_when_enabled » file_name

field file_enabled: Annotated[bool, FieldInfo(annotation=NoneType, required=False, default=False, description='Enable logging to file')] = False¶: Enable logging to file

field file_name: Annotated[str, FieldInfo(annotation=NoneType, required=False, default='/home/docs/.local/share/aignostics/aignostics.log', description='Name of the log file')] = '/home/docs/.local/share/aignostics/aignostics.log'¶

Name of the log file

Validated by:

validate_file_name_when_enabled

field level: Annotated[Literal['CRITICAL', 'ERROR', 'WARNING', 'SUCCESS', 'INFO', 'DEBUG', 'TRACE'], FieldInfo(annotation=NoneType, required=False, default='INFO', description='Log level, see https://loguru.readthedocs.io/en/stable/api/logger.html')] = 'INFO'¶: Log level, see https://loguru.readthedocs.io/en/stable/api/logger.html

field redirect_logging: Annotated[bool, FieldInfo(annotation=NoneType, required=False, default=False, description='Redirect standard logging')] = False¶: Redirect standard logging

field stderr_enabled: Annotated[bool, FieldInfo(annotation=NoneType, required=False, default=True, description='Enable logging to stderr')] = True¶: Enable logging to stderr

validator validate_file_name_when_enabled » file_name¶

Validate file_name only when file_enabled is True.

Parameters:

file_name – The file name to validate.
info – Validation info containing other field values.

Returns:

The validated file name.

Return type:

str

class aignostics.utils.NavGroup(name: str, icon: str = 'folder', items: list[~aignostics.utils._nav.NavItem] = <factory>, position: int = 1000, use_expansion: bool = True)¶

Group of navigation items from a NavBuilder.

Used internally for rendering navigation in the sidebar.

class aignostics.utils.NavItem(icon: str, label: str, target: str, marker: str | None = None, new_tab: bool = False)¶

Navigation item for sidebar.

icon¶

Material icon name (e.g., ‘waving_hand’, ‘settings’).

Type:: str

label¶

Display label for the navigation item.

Type:: str

target¶

URL path or external URL for the link.

Type:: str

marker¶

Test marker for the item. Auto-generated from label if None.

Type:: str | None

new_tab¶

Whether to open the link in a new tab. Defaults to False (same tab).

Type:: bool

__post_init__() → None¶: Auto-generate marker from label if not provided.

pydantic settings aignostics.utils.OpaqueSettings¶

Show JSON schema

{
   "title": "OpaqueSettings",
   "type": "object",
   "properties": {},
   "additionalProperties": false
}

static serialize_path_resolve(input_value: Path, _info: FieldSerializationInfo) → str | None¶

static serialize_sensitive_info(input_value: SecretStr, info: FieldSerializationInfo) → str | None¶

pydantic model aignostics.utils.ProcessInfo¶

Information about the current process.

Fields:

parent (aignostics.utils._process.ParentProcessInfo)
pid (int)
project_root (str)

field parent: ParentProcessInfo [Required]¶

field pid: int [Required]¶

field project_root: str [Required]¶

aignostics.utils.boot(sentry_integrations: list[Integration] | None, log_filter: Callable[[Any], bool] | None = None) → None¶

Boot the application or library.

Parameters:

sentry_integrations (list[Integration] | None) – List of Sentry integrations to use
log_filter (Callable[[Any], bool] | None) – Optional filter function for logging

aignostics.utils.create_marimo_app(notebook: Path, override_if_exists: bool) → Any¶

Create a FastAPI app with marimo notebook server.

Parameters:

notebook (Path) – Path to the notebook. Notebook will be copied into the user data directory.
override_if_exists (bool) – Whether to override the notebook in the user data directory if it already exists.

Returns:

FastAPI app with marimo notebook server.

Return type:

FastAPI

Raises:

ValueError – If the notebook directory does not exist.

aignostics.utils.discover_plugin_packages() → tuple[str, ...]¶

Discover plugin packages using entry points.

Plugins register themselves in their pyproject.toml:

[project.entry-points.”aignostics.plugins”] my_plugin = “my_plugin”

Results are cached after the first call.

Returns:: Tuple of discovered plugin package names.
Return type:: tuple[str, …]

aignostics.utils.get_process_info() → ProcessInfo¶

Get information about the current process and its parent.

Returns:: Object containing process information.
Return type:: ProcessInfo

aignostics.utils.get_user_data_directory(scope: str | None = None) → Path¶

Get the data directory for the service. Directory created if it does not exist.

Parameters:: scope (str | None) – Optional scope for the data directory.
Returns:: The data directory path.
Return type:: Path

aignostics.utils.gui_get_nav_groups() → list[NavGroup]¶

Collect navigation groups from all NavBuilders.

Returns:: Navigation groups sorted by position.
Return type:: list[NavGroup]

aignostics.utils.gui_register_pages() → None¶

This function is called by the GUI to register all pages.

aignostics.utils.gui_run(native: bool = True, show: bool = False, host: str | None = None, port: int | None = None, title: str = 'aignostics', icon: str = '', watch: bool = False, with_api: bool = False, dark_mode: bool = False) → None¶

Start the GUI.

Parameters:

native – Whether to run the GUI in native mode.
show – Whether to show the GUI.
host – Host to run the GUI on.
port – Port to run the GUI on.
title – Title of the GUI.
icon – Icon for the GUI.
watch – Whether to watch for changes and reload the GUI.
with_api – Whether to mount the API.
dark_mode – Whether to use dark mode.

Raises:

ValueError – If with_notebook is True but notebook_path is None, or trying to run native within container.

aignostics.utils.load_settings(settings_class: type[T]) → T¶

Load settings with error handling and nice formatting.

Parameters:

settings_class – The Pydantic settings class to instantiate

Returns:

Instance of the settings class

Return type:

Raises:

SystemExit – If settings validation fails

aignostics.utils.locate_implementations(_class: type[Any]) → list[Any]¶

Dynamically discover all instances of some class.

Searches plugin packages using a shallow scan and the main project package using a deep scan.

Parameters:: _class (type[Any]) – Class to search for.
Returns:: List of discovered implementations of the given class.
Return type:: list[Any]

aignostics.utils.locate_subclasses(_class: type[Any]) → list[Any]¶

Dynamically discover all classes that are subclasses of some type.

Searches plugin packages using a shallow scan and the main project package using a deep scan.

Parameters:: _class (type[Any]) – Parent class of subclasses to search for.
Returns:: List of discovered subclasses of the given class.
Return type:: list[type[Any]]

aignostics.utils.mcp_create_server(server_name: str = 'Central Aignostics MCP Server') → FastMCP¶

Create and configure the MCP server with all discovered plugins mounted.

Creates a new FastMCP server instance and mounts all discovered MCP servers from the SDK and plugins. Each mounted server’s tools are namespaced automatically using FastMCP’s built-in namespace feature.

Parameters:: server_name – Human-readable name for the MCP server.
Returns:: Configured MCP server ready to run.
Return type:: FastMCP

aignostics.utils.mcp_discover_servers() → list[FastMCP]¶

Discover all FastMCP server instances from SDK and plugins.

Uses a plugin discovery mechanism which: - Searches the main aignostics package - Searches all registered plugin packages via entry points

Returns:: List of discovered FastMCP server instances.
Return type:: list[FastMCP]

aignostics.utils.mcp_list_tools(server_name: str = 'Central Aignostics MCP Server') → list[dict[str, Any]]¶

List all available MCP tools.

Creates the server and returns information about all registered tools including those from mounted servers.

Note

This function must be called from a synchronous context. Calling it from within an async function will raise RuntimeError.

Parameters:

server_name – Human-readable name for the MCP server.

Returns:

List of tool information dictionaries with: ’name’ and ‘description’ keys.

Return type:

list[dict[str, Any]]

aignostics.utils.mcp_run(server_name: str = 'Central Aignostics MCP Server') → None¶

Run the MCP server using stdio transport.

Starts an MCP server that exposes SDK functionality to AI agents. The server automatically discovers and mounts tools from the SDK and any installed plugins.

Parameters:: server_name – Human-readable name for the MCP server.

aignostics.utils.open_user_data_directory(scope: str | None = None) → Path¶

Open the user data directory in the file manager of the respective system platform.

Parameters:: scope (str | None) – Optional scope for the data directory.
Returns:: The data directory path.
Return type:: Path

aignostics.utils.prepare_cli(cli: Typer, epilog: str) → None¶

Dynamically locate, register and prepare subcommands.

Parameters:

cli (typer.Typer) – Typer instance
epilog (str) – Epilog to add

aignostics.utils.sanitize_path(path: str | Path) → str | Path¶

Sanitize a filesystem path.

Rules: 1. If a Path is provided a Path will returned, otherwise a string will be returned. 2. Colons will be replaced with underscores except if it’s a Windows drive letter. 3. On Windows: If the sanitized path is reserved a ValueError will be raised.

Parameters:: path (str | Path) – The path to sanitize.
Returns:: The sanitized path.
Return type:: str | Path
Raises:: ValueError – If the sanitized path is reserved on Windows.

aignostics.utils.sanitize_path_component(component: str) → str¶

Sanitize a single path component (e.g., filename, directory name).

Rules: 1. Replaces all colons with underscores (no drive letter exceptions).

This function is useful for sanitizing individual path components like filepath.stem where you don’t want drive letter special handling.

Parameters:: component (str) – The path component to sanitize.
Returns:: The sanitized path component.
Return type:: str

aignostics.utils.user_agent() → str¶

Generate a user agent string for HTTP requests.

Format: {project_name}/{version} ({platform}; {current_test}; {github_run})

Returns:: The user agent string.
Return type:: str

Python SDK providing access to Aignostics AI services.