AwsSession

class braket.aws.aws_session.AwsSession(boto_session=None, braket_client=None, config=None, default_bucket=None)[source]

Bases: object

Manage interactions with AWS services.

Parameters:

boto_session (boto3.Session | None)
braket_client (client | None)
config (Config | None)
default_bucket (str | None)

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: <module 'botocore.client' from '/home/docs/checkouts/readthedocs.org/user_builds/amazon-braket-sdk-python/envs/1220/lib/python3.13/site-packages/botocore/client.py'>

Gets the IAM client.

Returns:: client – The IAM Client.

property s3_client: <module 'botocore.client' from '/home/docs/checkouts/readthedocs.org/user_builds/amazon-braket-sdk-python/envs/1220/lib/python3.13/site-packages/botocore/client.py'>

Gets the S3 client.

Returns:: client – The S3 Client.

property sts_client: <module 'botocore.client' from '/home/docs/checkouts/readthedocs.org/user_builds/amazon-braket-sdk-python/envs/1220/lib/python3.13/site-packages/botocore/client.py'>

Gets the STS client.

Returns:: client – The STS Client.

property logs_client: <module 'botocore.client' from '/home/docs/checkouts/readthedocs.org/user_builds/amazon-braket-sdk-python/envs/1220/lib/python3.13/site-packages/botocore/client.py'>

Gets the CloudWatch logs client.

Returns:: client – The CloudWatch logs Client.

property ecr_client: <module 'botocore.client' from '/home/docs/checkouts/readthedocs.org/user_builds/amazon-braket-sdk-python/envs/1220/lib/python3.13/site-packages/botocore/client.py'>

Gets the ECR client.

Returns:: client – The ECR Client.

add_braket_user_agent(user_agent)[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.
Return type:: None

cancel_quantum_task(arn)[source]

Cancel the quantum task.

Parameters:: arn (str) – The ARN of the quantum task to cancel.
Return type:: None

create_quantum_task(**boto3_kwargs)[source]

Create a quantum task.

Parameters:: **boto3_kwargs – Keyword arguments for the Amazon Braket CreateQuantumTask operation.
Return type:: str
Returns:: str – The ARN of the quantum task.

create_job(**boto3_kwargs)[source]

Create a quantum hybrid job.

Parameters:: **boto3_kwargs – Keyword arguments for the Amazon Braket CreateJob operation.
Return type:: str
Returns:: str – The ARN of the hybrid job.

get_quantum_task(arn)[source]

Gets the quantum task.

Parameters:: arn (str) – The ARN of the quantum task to get.
Return type:: dict[str, Any]
Returns:: dict[str, Any] – The response from the Amazon Braket GetQuantumTask operation.

get_default_jobs_role()[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/.

Return type:: str
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)[source]

Gets the hybrid job.

Parameters:: arn (str) – The ARN of the hybrid job to get.
Return type:: dict[str, Any]
Returns:: dict[str, Any] – The response from the Amazon Braket GetQuantumJob operation.

cancel_job(arn)[source]

Cancel the hybrid job.

Parameters:: arn (str) – The ARN of the hybrid job to cancel.
Return type:: dict[str, Any]
Returns:: dict[str, Any] – The response from the Amazon Braket CancelJob operation.

retrieve_s3_object_body(s3_bucket, s3_object_key)[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.

Return type:

str

Returns:

str – The body of the S3 object.

upload_to_s3(filename, s3_uri)[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.

Return type:

None

upload_local_data(local_prefix, s3_prefix)[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-‘)

Return type:

None

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, filename)[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.

Return type:

None

copy_s3_object(source_s3_uri, destination_s3_uri)[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.

Return type:

None

copy_s3_directory(source_s3_path, destination_s3_path)[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.

Return type:

None

list_keys(bucket, prefix)[source]

Lists keys matching prefix in bucket.

Parameters:

bucket (str) – Bucket to be queried.
prefix (str) – The S3 path prefix to be matched

Return type:

list[str]

Returns:

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

default_bucket()[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.

Return type:: str
Returns:: str – Name of the default bucket.

get_device(arn)[source]

Calls the Amazon Braket get_device API to retrieve device metadata.

Parameters:: arn (str) – The ARN of the device.
Return type:: dict[str, Any]
Returns:: dict[str, Any] – The response from the Amazon Braket GetDevice operation.

search_devices(arns=None, names=None, types=None, statuses=None, provider_names=None)[source]

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

Parameters:

arns (list[str] | None) – device ARN filter, default is None.
names (list[str] | None) – device name filter, default is None.
types (list[str] | None) – device type filter, default is None.
statuses (list[str] | None) – 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 (list[str] | None) – provider name list, default is None.

Return type:

list[dict[str, Any]]

Returns:

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

static is_s3_uri(string)[source]

Determines if a given string is an S3 URI.

Parameters:: string (str) – the string to check.
Return type:: bool
Returns:: bool – Returns True if the given string is an S3 URI.

static parse_s3_uri(s3_uri)[source]

Parse S3 URI to get bucket and key

Parameters:

s3_uri (str) – S3 URI.

Return type:

tuple[str, str]

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, *dirs)[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

Return type:

str

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, log_stream_prefix, limit=None, next_token=None)[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 (int | None) – Limit for number of log streams returned. default is 50.
next_token (str | None) – The token for the next set of items to return. Would have been received in a previous call.

Return type:

dict[str, Any]

Returns:

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

get_log_events(log_group, log_stream, start_time, start_from_head=True, next_token=None)[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 (str | None) – The token for the next set of items to return. Would have been received in a previous call.

Return type:

dict[str, Any]

Returns:

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

copy_session(region=None, max_connections=None)[source]

Creates a new AwsSession based on the region.

Parameters:

region (str | None) – Name of the region. Default = None.
max_connections (int | None) – The maximum number of connections in the Boto3 connection pool. Default = None.

Return type:

AwsSession

Returns:

AwsSession – based on the region and boto config parameters.

get_full_image_tag(image_uri)[source]

Get verbose image tag from image uri.

Parameters:: image_uri (str) – Image uri to get tag for.
Return type:: str
Returns:: str – Verbose image tag for given image.