roboto.domain.actions.trigger_operations#

Module Contents#

class roboto.domain.actions.trigger_operations.CreateTriggerRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload to create a new trigger.

Contains all the configuration needed to create a trigger that automatically invokes actions when specific conditions are met.

Parameters:: data (Any)

action_digest: str | None = None#: Optional specific version digest of the action to invoke. If not provided, uses the latest version.

action_name: str#: Name of the action to invoke when the trigger fires.

action_owner_id: str | None = None#: Organization ID that owns the target action. If not provided, searches in the caller’s organization.

additional_inputs: list[str] | None = None#: Optional additional file patterns to include in action invocations beyond the required inputs.

causes: list[roboto.domain.actions.trigger_record.TriggerEvaluationCause] | None = None#: List of events that can cause this trigger to be evaluated. If not provided, uses default causes.

compute_requirement_overrides: roboto.domain.actions.ComputeRequirements | None = None#: Optional compute requirement overrides for action invocations.

condition: roboto.query.ConditionType | None = None#

Optional condition that must be met for the trigger to fire.

Can filter based on metadata, file properties, etc.

container_parameter_overrides: roboto.domain.actions.ContainerParameters | None = None#: Optional container parameter overrides for action invocations.

enabled: bool = True#: Whether the trigger should be active immediately after creation.

for_each: roboto.domain.actions.trigger_record.TriggerForEachPrimitive#: Granularity of execution - Dataset or DatasetFile.

name: str = None#: Unique name for the trigger (alphanumeric, hyphens, underscores only, max 256 characters).

parameter_values: dict[str, Any] | None = None#: Parameter values to pass to the action when invoked.

required_inputs: list[str]#: List of file patterns that must be present for the trigger to fire. Uses glob patterns like ‘**/*.bag’.

service_user_id: str | None = None#: Optional service user ID for authentication.

timeout: int | None = None#: Optional timeout override for action invocations in minutes.

validate_additional_inputs(value)#

Parameters:: value (Optional[list[str]])
Return type:: Optional[list[str]]

validate_required_inputs(value)#

Parameters:: value (list[str])
Return type:: list[str]

class roboto.domain.actions.trigger_operations.EvaluateTriggersRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload to manually evaluate specific triggers.

Used to force evaluation of triggers outside of their normal automatic evaluation cycle. This is typically used for testing or debugging trigger behavior.

Note

This is primarily used internally by the Roboto platform and is not commonly needed by SDK users.

Parameters:: data (Any)

trigger_evaluation_ids: collections.abc.Iterable[int]#: Collection of trigger evaluation IDs to process.

class roboto.domain.actions.trigger_operations.QueryTriggersRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload to query triggers with filters.

Used to search for triggers based on various criteria such as name, status, or other attributes.

Parameters:: data (Any)

filters: dict[str, Any] = None#: Dictionary of filter criteria to apply to the trigger search.

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class roboto.domain.actions.trigger_operations.TriggerEvaluationsSummaryResponse(/, **data)#

Bases: pydantic.BaseModel

Response containing summary information about trigger evaluations.

Provides high-level statistics about trigger evaluation status, useful for monitoring and debugging trigger performance.

Parameters:: data (Any)

count_pending: int#: Number of trigger evaluations currently pending.

last_evaluation_start: datetime.datetime | None#: Timestamp of the most recent evaluation start, if any evaluations have occurred.

class roboto.domain.actions.trigger_operations.UpdateTriggerRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload to update an existing trigger.

Contains the changes to apply to a trigger. Only specified fields will be updated; others remain unchanged. Uses NotSet sentinel values to distinguish between explicit None values and unspecified fields.

Parameters:: data (Any)

action_digest: str | roboto.sentinels.NotSetType | None#: New specific version digest of the action.

action_name: str | roboto.sentinels.NotSetType#: New action name to invoke.

action_owner_id: str | roboto.sentinels.NotSetType#: New organization ID that owns the target action.

additional_inputs: list[str] | roboto.sentinels.NotSetType | None#: New additional file patterns to include.

causes: list[roboto.domain.actions.trigger_record.TriggerEvaluationCause] | roboto.sentinels.NotSetType#: New list of events that can cause trigger evaluation.

compute_requirement_overrides: roboto.domain.actions.ComputeRequirements | roboto.sentinels.NotSetType | None#: New compute requirement overrides.

condition: roboto.query.ConditionType | roboto.sentinels.NotSetType | None#: New condition that must be met for trigger to fire.

container_parameter_overrides: roboto.domain.actions.ContainerParameters | roboto.sentinels.NotSetType | None#: New container parameter overrides.

enabled: bool | roboto.sentinels.NotSetType#: New enabled status for the trigger.

for_each: roboto.domain.actions.trigger_record.TriggerForEachPrimitive | roboto.sentinels.NotSetType#: New execution granularity (Dataset or DatasetFile).

model_config#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

parameter_values: dict[str, Any] | roboto.sentinels.NotSetType | None#: New parameter values to pass to the action.

required_inputs: list[str] | roboto.sentinels.NotSetType#: New list of required file patterns.

timeout: int | roboto.sentinels.NotSetType | None#: New timeout override for action invocations.