braket.aws.aws_session module

class braket.aws.aws_session.AwsSession(boto_session: boto3.Session | None = None, braket_client: client | None = None, config: Config | None = None, default_bucket: str | None = None)[source]

Bases: object

Manage interactions with AWS services.

Initializes an AwsSession.

Parameters:

  • boto_session (boto3.Session | None) – A boto3 session object.

  • braket_client (client | None) – A boto3 Braket client.

  • config (Config | None) – A botocore Config object.

  • default_bucket (str | None) – The name of the default bucket of the AWS Session.

Raises:

ValueError – invalid boto_session or braket_client.

class S3DestinationFolder(bucket: str, key: str)[source]

Bases: NamedTuple

A NamedTuple for an S3 bucket and object key.

Create new instance of S3DestinationFolder(bucket, key)

bucket: str

Alias for field number 0

key: str

Alias for field number 1

property region: str
property account_id: str

Gets the caller’s account number.

Returns:

str – The account number of the caller.

property iam_client: client

Gets the IAM client.

Returns:

client – The IAM Client.

property s3_client: client

Gets the S3 client.

Returns:

client – The S3 Client.

property sts_client: client

Gets the STS client.

Returns:

client – The STS Client.

property logs_client: client

Gets the CloudWatch logs client.

Returns:

client – The CloudWatch logs Client.

property ecr_client: client

Gets the ECR client.

Returns:

client – The ECR Client.

add_braket_user_agent(user_agent: str) → None[source]

Appends the user-agent value to the User-Agent header, if it does not yet exist in the header. This method is typically only relevant for libraries integrating with the Amazon Braket SDK.

Parameters:

user_agent (str) – The user_agent value to append to the header.

cancel_quantum_task(arn: str) → None[source]

Cancel the quantum task.

Parameters:

arn (str) – The ARN of the quantum task to cancel.

create_quantum_task(**boto3_kwargs) → str[source]

Create a quantum task.

Parameters:

**boto3_kwargs – Keyword arguments for the Amazon Braket CreateQuantumTask operation.

Returns:

str – The ARN of the quantum task.

create_job(**boto3_kwargs) → str[source]

Create a quantum hybrid job.

Parameters:

**boto3_kwargs – Keyword arguments for the Amazon Braket CreateJob operation.

Returns:

str – The ARN of the hybrid job.

get_quantum_task(arn: str) → dict[str, Any][source]

Gets the quantum task.

Parameters:

arn (str) – The ARN of the quantum task to get.

Returns:

dict[str, Any] – The response from the Amazon Braket GetQuantumTask operation.

get_default_jobs_role() → str[source]

This returns the role ARN for the default hybrid jobs role created in the Amazon Braket Console. It will pick the first role it finds with the RoleName prefix AmazonBraketJobsExecutionRole with a PathPrefix of /service-role/.

Returns:

str – The ARN for the default IAM role for jobs execution created in the Amazon Braket console.

Raises:

RuntimeError – If no roles can be found with the prefix /service-role/AmazonBraketJobsExecutionRole.

get_job(arn: str) → dict[str, Any][source]

Gets the hybrid job.

Parameters:

arn (str) – The ARN of the hybrid job to get.

Returns:

dict[str, Any] – The response from the Amazon Braket GetQuantumJob operation.

cancel_job(arn: str) → dict[str, Any][source]

Cancel the hybrid job.

Parameters:

arn (str) – The ARN of the hybrid job to cancel.

Returns:

dict[str, Any] – The response from the Amazon Braket CancelJob operation.

retrieve_s3_object_body(s3_bucket: str, s3_object_key: str) → str[source]

Retrieve the S3 object body.

Parameters:

  • s3_bucket (str) – The S3 bucket name.

  • s3_object_key (str) – The S3 object key within the s3_bucket.

Returns:

str – The body of the S3 object.

upload_to_s3(filename: str, s3_uri: str) → None[source]

Upload file to S3.

Parameters:

  • filename (str) – local file to be uploaded.

  • s3_uri (str) – The S3 URI where the file will be uploaded.

upload_local_data(local_prefix: str, s3_prefix: str) → None[source]

Upload local data matching a prefix to a corresponding location in S3

Parameters:

  • local_prefix (str) – a prefix designating files to be uploaded to S3. All files beginning with local_prefix will be uploaded.

  • s3_prefix (str) – the corresponding S3 prefix that will replace the local prefix when the data is uploaded. This will be an S3 URI and should include the bucket (i.e. ‘s3://my-bucket/my/prefix-‘)

Example

local_prefix = “input”, s3_prefix = “s3://my-bucket/dir/input” will upload:

  • ‘input.csv’ to ‘s3://my-bucket/dir/input.csv’

  • ‘input-2.csv’ to ‘s3://my-bucket/dir/input-2.csv’

  • ‘input/data.txt’ to ‘s3://my-bucket/dir/input/data.txt’

  • ‘input-dir/data.csv’ to ‘s3://my-bucket/dir/input-dir/data.csv’ but will not upload:

  • ‘my-input.csv’

  • ‘my-dir/input.csv’

To match all files within the directory “input” and upload them into

“s3://my-bucket/input”, provide local_prefix = “input/” and s3_prefix = “s3://my-bucket/input/”

download_from_s3(s3_uri: str, filename: str) → None[source]

Download file from S3

Parameters:

  • s3_uri (str) – The S3 uri from where the file will be downloaded.

  • filename (str) – filename to save the file to.

copy_s3_object(source_s3_uri: str, destination_s3_uri: str) → None[source]

Copy object from another location in s3. Does nothing if source and destination URIs are the same.

Parameters:

  • source_s3_uri (str) – S3 URI pointing to the object to be copied.

  • destination_s3_uri (str) – S3 URI where the object will be copied to.

copy_s3_directory(source_s3_path: str, destination_s3_path: str) → None[source]

Copy all objects from a specified directory in S3. Does nothing if source and destination URIs are the same. Preserves nesting structure, will not overwrite other files in the destination location unless they share a name with a file being copied.

Parameters:

  • source_s3_path (str) – S3 URI pointing to the directory to be copied.

  • destination_s3_path (str) – S3 URI where the contents of the source_s3_path directory will be copied to.

list_keys(bucket: str, prefix: str) → list[str][source]

Lists keys matching prefix in bucket.

Parameters:

  • bucket (str) – Bucket to be queried.

  • prefix (str) – The S3 path prefix to be matched

Returns:

list[str] – A list of all keys matching the prefix in the bucket.

default_bucket() → str[source]

Returns the name of the default bucket of the AWS Session. In the following order of priority, it will return either the parameter default_bucket set during initialization of the AwsSession (if not None), the bucket being used by the currently running Braket Hybrid Job (if evoked inside of a Braket Hybrid Job), or a default value of “amazon-braket-<aws account id>-<aws session region>. Except in the case of a user- specified bucket name, this method will create the default bucket if it does not exist.

Returns:

str – Name of the default bucket.

get_device(arn: str) → dict[str, Any][source]

Calls the Amazon Braket get_device API to retrieve device metadata.

Parameters:

arn (str) – The ARN of the device.

Returns:

dict[str, Any] – The response from the Amazon Braket GetDevice operation.

search_devices(arns: list[str] | None = None, names: list[str] | None = None, types: list[str] | None = None, statuses: list[str] | None = None, provider_names: list[str] | None = None) → list[dict[str, Any]][source]

Get devices based on filters. The result is the AND of all the filters arns, names, types, statuses, provider_names.

Parameters:

  • arns (Optional[list[str]]) – device ARN filter, default is None.

  • names (Optional[list[str]]) – device name filter, default is None.

  • types (Optional[list[str]]) – device type filter, default is None.

  • statuses (Optional[list[str]]) – device status filter, default is None. When None is used, RETIRED devices will not be returned. To include RETIRED devices in the results, use a filter that includes “RETIRED” for this parameter.

  • provider_names (Optional[list[str]]) – provider name list, default is None.

Returns:

list[dict[str, Any]] – The response from the Amazon Braket SearchDevices operation.

static is_s3_uri(string: str) → bool[source]

Determines if a given string is an S3 URI.

Parameters:

string (str) – the string to check.

Returns:

bool – Returns True if the given string is an S3 URI.

static parse_s3_uri(s3_uri: str) → tuple[str, str][source]

Parse S3 URI to get bucket and key

Parameters:

s3_uri (str) – S3 URI.

Returns:

tuple[str, str] – Bucket and Key tuple.

Raises:

  • ValueError – Raises a ValueError if the provided string is not

  • a valid S3 URI. –

static construct_s3_uri(bucket: str, *dirs: str) → str[source]

Create an S3 URI given a bucket and path.

Parameters:

  • bucket (str) – S3 URI.

  • *dirs (str) – directories to be appended in the resulting S3 URI

Returns:

str – S3 URI

Raises:

  • ValueError – Raises a ValueError if the provided arguments are not

  • valid to generate an S3 URI –

describe_log_streams(log_group: str, log_stream_prefix: str, limit: int | None = None, next_token: str | None = None) → dict[str, Any][source]

Describes CloudWatch log streams in a log group with a given prefix.

Parameters:

  • log_group (str) – Name of the log group.

  • log_stream_prefix (str) – Prefix for log streams to include.

  • limit (Optional[int]) – Limit for number of log streams returned. default is 50.

  • next_token (Optional[str]) – The token for the next set of items to return. Would have been received in a previous call.

Returns:

dict[str, Any] – Dictionary containing logStreams and nextToken

get_log_events(log_group: str, log_stream: str, start_time: int, start_from_head: bool = True, next_token: str | None = None) → dict[str, Any][source]

Gets CloudWatch log events from a given log stream.

Parameters:

  • log_group (str) – Name of the log group.

  • log_stream (str) – Name of the log stream.

  • start_time (int) – Timestamp that indicates a start time to include log events.

  • start_from_head (bool) – Bool indicating to return oldest events first. default is True.

  • next_token (Optional[str]) – The token for the next set of items to return. Would have been received in a previous call.

Returns:

dict[str, Any] – Dictionary containing events, nextForwardToken, and nextBackwardToken

copy_session(region: str | None = None, max_connections: int | None = None) → AwsSession[source]

Creates a new AwsSession based on the region.

Parameters:

  • region (Optional[str]) – Name of the region. Default = None.

  • max_connections (Optional[int]) – The maximum number of connections in the Boto3 connection pool. Default = None.

Returns:

AwsSession – based on the region and boto config parameters.

get_full_image_tag(image_uri: str) → str[source]

Get verbose image tag from image uri.

Parameters:

image_uri (str) – Image uri to get tag for.

Returns:

str – Verbose image tag for given image.