roboto.domain.orgs.s3_integration

Module Contents

RegisterS3IntegrationRequest

class roboto.domain.orgs.s3_integration.RegisterS3IntegrationRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload to integrate an S3 bucket with Roboto.

Parameters

data Any

Attributes

RegisterS3IntegrationRequest.account_id

account_id str #

AWS account ID that owns the S3 bucket.

RegisterS3IntegrationRequest.aws_region

aws_region str #

AWS region where the S3 bucket is located.

RegisterS3IntegrationRequest.bucket_name

bucket_name str #

Name of the S3 bucket to integrate.

RegisterS3IntegrationRequest.org_id

org_id str #

Organization ID to associate with this S3 integration.

RegisterS3IntegrationRequest.readonly

readonly bool = False #

Whether Roboto should have read-only access to the bucket.

RegisterS3IntegrationRequest.transfer_accelerated

transfer_accelerated bool = False #

Whether to enable S3 Transfer Acceleration for faster uploads.

RegisterS3IntegrationResponse

class roboto.domain.orgs.s3_integration.RegisterS3IntegrationResponse(/, **data)#

Bases: pydantic.BaseModel

Response payload containing S3 integration setup instructions.

Parameters

data Any

Attributes

RegisterS3IntegrationResponse.iam_role_name

iam_role_name str #

Name of the IAM role to create for Roboto access.

RegisterS3IntegrationResponse.iam_role_policy

iam_role_policy dict[str, Any] #

IAM policy document to attach to the role.

RegisterS3IntegrationResponse.iam_role_trust_relationship

iam_role_trust_relationship dict[str, Any] #

IAM trust policy document for the role.

RegisterS3IntegrationResponse.s3_bucket_cors_policy

s3_bucket_cors_policy list[dict[str, Any]] #

CORS policy to apply to the S3 bucket.

S3BucketHealthCheck

class roboto.domain.orgs.s3_integration.S3BucketHealthCheck(/, **data)#

Bases: pydantic.BaseModel

Summary of the most recent health check against an S3 bucket integration.

Carries only a machine-readable classification and curated, non-sensitive copy: no identifiers (bucket name, role ARN, account ID), no secrets, and no raw AWS error text. The identifiers a reader needs live on the enclosing S3BucketIntegrationRecord.

Parameters

data Any

Attributes

S3BucketHealthCheck.aws_error_code

aws_error_code str | None = None #

AWS error code that drove the classification (e.g. AccessDenied), if any — a fixed token, never AWS’s free-text message.

S3BucketHealthCheck.message

message str #

Curated, human-readable explanation, safe to display.

S3BucketHealthCheck.probes

probes dict[str, str] = None #

Outcome per probe name (assume_role, list, put, …), each one of ok / denied / error / skipped.

S3BucketHealthCheck.reason_code

reason_code str #

Machine-readable result classification — a fixed token, e.g. ok or permission_denied.

S3BucketIntegrationRecord

class roboto.domain.orgs.s3_integration.S3BucketIntegrationRecord(/, **data)#

Bases: pydantic.BaseModel

Record representing an S3 bucket integration with a Roboto organization.

Parameters

data Any

Attributes

S3BucketIntegrationRecord.aws_region

aws_region str #

AWS region where the S3 bucket is located.

S3BucketIntegrationRecord.bucket_name

bucket_name str #

Name of the integrated S3 bucket.

S3BucketIntegrationRecord.created

created datetime.datetime #

Timestamp when the integration was created.

S3BucketIntegrationRecord.last_health_check

last_health_check S3BucketHealthCheck | None = None #

Summary of the most recent health check, or None if the integration has never been checked.

S3BucketIntegrationRecord.modified

modified datetime.datetime #

Timestamp when the integration was last modified.

S3BucketIntegrationRecord.org_id

org_id str #

Organization ID this bucket is associated with.

S3BucketIntegrationRecord.readonly

readonly bool #

Whether Roboto has read-only access to the bucket.

S3BucketIntegrationRecord.status

status str #

Integration status: one of ‘unverified’, ‘healthy’, ‘unhealthy’, ‘unknown’.

S3BucketIntegrationRecord.status_last_updated

status_last_updated datetime.datetime | None = None #

Timestamp when status was last written, i.e. when the most recent health check ran.

S3IntegrationService

class roboto.domain.orgs.s3_integration.S3IntegrationService(roboto_client, sts_client=None, s3_client=None, iam_client=None)#

Service for integrating S3 buckets with Roboto organizations.

This service handles the setup of cross-account IAM roles and S3 bucket policies to allow Roboto to access customer S3 buckets for data storage and processing.

Parameters

roboto_client roboto.http.RobotoClient

sts_client Optional[Any]

s3_client Optional[Any]

iam_client Optional[Any]

S3IntegrationService.register_bucket()

register_bucket(org_id, account_id, bucket_name, transfer_accelerated=False, readonly=False)#

This method sets up the necessary IAM roles and S3 bucket policies to allow Roboto to access the specified S3 bucket. The caller must have appropriate AWS credentials with permissions to create IAM roles and modify S3 bucket policies.

Parameters

org_id str

Organization ID to associate with this S3 integration.

account_id str

AWS account ID that owns the S3 bucket.

bucket_name str

Name of the S3 bucket to integrate.

transfer_accelerated bool

Whether to enable S3 Transfer Acceleration.

readonly bool

Whether Roboto should have read-only access to the bucket.

Raises

RobotoInvalidRequestException

AWS credentials are invalid or account ID mismatch.

ValueError

The specified bucket does not exist or is not owned by the account.

botocore.exceptions.ClientError

AWS API errors during setup.

Usage

from roboto.domain.orgs import S3IntegrationService
from roboto import RobotoClient
service = S3IntegrationService(RobotoClient())
service.register_bucket(org_id="org_12345", account_id="123456789012", bucket_name="my-data-bucket")

service.register_bucket(
    org_id="org_12345", account_id="123456789012", bucket_name="my-readonly-bucket", readonly=True
)

logger

roboto.domain.orgs.s3_integration.logger#