Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _resource_arn method in localstack

Best Python code snippet using localstack_python

aws_stack.py

Source:aws_stack.py

...501    if ":policy/" in policy_name:502        return policy_name503    account_id = account_id or get_aws_account_id()504    return "arn:aws:iam::{}:policy/{}".format(account_id, policy_name)505def iam_resource_arn(resource, role=None, env=None):506    env = get_environment(env)507    if not role:508        role = get_iam_role(resource, env=env)509    return role_arn(role_name=role, account_id=get_aws_account_id())510def get_iam_role(resource, env=None):511    env = get_environment(env)512    return "role-%s" % resource513def secretsmanager_secret_arn(secret_id, account_id=None, region_name=None, random_suffix=None):514    if ":" in (secret_id or ""):515        return secret_id516    pattern = "arn:aws:secretsmanager:%s:%s:secret:%s"517    arn = _resource_arn(secret_id, pattern, account_id=account_id, region_name=region_name)518    if random_suffix:519        arn += f"-{random_suffix}"520    return arn521def cloudformation_stack_arn(stack_name, stack_id=None, account_id=None, region_name=None):522    stack_id = stack_id or "id-123"523    pattern = "arn:aws:cloudformation:%s:%s:stack/%s/{stack_id}".format(stack_id=stack_id)524    return _resource_arn(stack_name, pattern, account_id=account_id, region_name=region_name)525def cf_change_set_arn(change_set_name, change_set_id=None, account_id=None, region_name=None):526    change_set_id = change_set_id or "id-456"527    pattern = "arn:aws:cloudformation:%s:%s:changeSet/%s/{cs_id}".format(cs_id=change_set_id)528    return _resource_arn(change_set_name, pattern, account_id=account_id, region_name=region_name)529def dynamodb_table_arn(table_name, account_id=None, region_name=None):530    table_name = table_name.split(":table/")[-1]531    pattern = "arn:aws:dynamodb:%s:%s:table/%s"532    return _resource_arn(table_name, pattern, account_id=account_id, region_name=region_name)533def dynamodb_stream_arn(table_name, latest_stream_label, account_id=None):534    account_id = account_id or get_aws_account_id()535    return "arn:aws:dynamodb:%s:%s:table/%s/stream/%s" % (536        get_region(),537        account_id,538        table_name,539        latest_stream_label,540    )541def cloudwatch_alarm_arn(alarm_name, account_id=None, region_name=None):542    pattern = "arn:aws:cloudwatch:%s:%s:alarm:%s"543    return _resource_arn(alarm_name, pattern, account_id=account_id, region_name=region_name)544def log_group_arn(group_name, account_id=None, region_name=None):545    pattern = "arn:aws:logs:%s:%s:log-group:%s"546    return _resource_arn(group_name, pattern, account_id=account_id, region_name=region_name)547def events_rule_arn(rule_name, account_id=None, region_name=None):548    pattern = "arn:aws:events:%s:%s:rule/%s"549    return _resource_arn(rule_name, pattern, account_id=account_id, region_name=region_name)550def lambda_function_arn(function_name, account_id=None, region_name=None):551    return lambda_function_or_layer_arn(552        "function", function_name, account_id=account_id, region_name=region_name553    )554def lambda_layer_arn(layer_name, version=None, account_id=None):555    return lambda_function_or_layer_arn("layer", layer_name, version=None, account_id=account_id)556def lambda_function_or_layer_arn(557    type, entity_name, version=None, account_id=None, region_name=None558):559    pattern = "arn:([a-z-]+):lambda:.*:.*:(function|layer):.*"560    if re.match(pattern, entity_name):561        return entity_name562    if ":" in entity_name:563        client = connect_to_service("lambda")564        entity_name, _, alias = entity_name.rpartition(":")565        try:566            alias_response = client.get_alias(FunctionName=entity_name, Name=alias)567            version = alias_response["FunctionVersion"]568        except Exception as e:569            msg = f"Alias {alias} of {entity_name} not found"570            LOG.info(f"{msg}: {e}")571            raise Exception(msg)572    account_id = account_id or get_aws_account_id()573    region_name = region_name or get_region()574    result = f"arn:aws:lambda:{region_name}:{account_id}:{type}:{entity_name}"575    if version:576        result = f"{result}:{version}"577    return result578def lambda_function_name(name_or_arn):579    if ":" in name_or_arn:580        arn = parse_arn(name_or_arn)581        if arn["service"] != "lambda":582            raise ValueError("arn is not a lambda arn %s" % name_or_arn)583        return parse_arn(name_or_arn)["resource"].split(":")[1]584    else:585        return name_or_arn586def state_machine_arn(name, account_id=None, region_name=None):587    pattern = "arn:aws:states:%s:%s:stateMachine:%s"588    return _resource_arn(name, pattern, account_id=account_id, region_name=region_name)589def stepfunctions_activity_arn(name, account_id=None, region_name=None):590    pattern = "arn:aws:states:%s:%s:activity:%s"591    return _resource_arn(name, pattern, account_id=account_id, region_name=region_name)592def fix_arn(arn):593    """Function that attempts to "canonicalize" the given ARN. This includes converting594    resource names to ARNs, replacing incorrect regions, account IDs, etc."""595    if arn.startswith("arn:aws:lambda"):596        parts = arn.split(":")597        region = parts[3] if parts[3] in get_valid_regions() else get_region()598        return lambda_function_arn(lambda_function_name(arn), region_name=region)599    LOG.warning("Unable to fix/canonicalize ARN: %s", arn)600    return arn601def cognito_user_pool_arn(user_pool_id, account_id=None, region_name=None):602    pattern = "arn:aws:cognito-idp:%s:%s:userpool/%s"603    return _resource_arn(user_pool_id, pattern, account_id=account_id, region_name=region_name)604def kinesis_stream_arn(stream_name, account_id=None, region_name=None):605    pattern = "arn:aws:kinesis:%s:%s:stream/%s"606    return _resource_arn(stream_name, pattern, account_id=account_id, region_name=region_name)607def elasticsearch_domain_arn(domain_name, account_id=None, region_name=None):608    pattern = "arn:aws:es:%s:%s:domain/%s"609    return _resource_arn(domain_name, pattern, account_id=account_id, region_name=region_name)610def firehose_stream_arn(stream_name, account_id=None, region_name=None):611    pattern = "arn:aws:firehose:%s:%s:deliverystream/%s"612    return _resource_arn(stream_name, pattern, account_id=account_id, region_name=region_name)613def es_domain_arn(domain_name, account_id=None, region_name=None):614    pattern = "arn:aws:es:%s:%s:domain/%s"615    return _resource_arn(domain_name, pattern, account_id=account_id, region_name=region_name)616def kms_key_arn(key_id: str, account_id: str = None, region_name: str = None) -> str:617    pattern = "arn:aws:kms:%s:%s:key/%s"618    return _resource_arn(key_id, pattern, account_id=account_id, region_name=region_name)619def kms_alias_arn(alias_name: str, account_id: str = None, region_name: str = None):620    if not alias_name.startswith("alias/"):621        alias_name = "alias/" + alias_name622    pattern = "arn:aws:kms:%s:%s:%s"623    return _resource_arn(alias_name, pattern, account_id=account_id, region_name=region_name)624def code_signing_arn(code_signing_id: str, account_id: str = None, region_name: str = None) -> str:625    pattern = "arn:aws:lambda:%s:%s:code-signing-config:%s"626    return _resource_arn(code_signing_id, pattern, account_id=account_id, region_name=region_name)627def ssm_parameter_arn(param_name: str, account_id: str = None, region_name: str = None) -> str:628    pattern = "arn:aws:ssm:%s:%s:parameter/%s"629    param_name = param_name.lstrip("/")630    return _resource_arn(param_name, pattern, account_id=account_id, region_name=region_name)631def s3_bucket_arn(bucket_name_or_arn: str, account_id=None):632    bucket_name = s3_bucket_name(bucket_name_or_arn)633    return "arn:aws:s3:::%s" % bucket_name634def s3_bucket_name(bucket_name_or_arn: str) -> str:635    return bucket_name_or_arn.split(":::")[-1]636def _resource_arn(name: str, pattern: str, account_id: str = None, region_name: str = None) -> str:637    if ":" in name:638        return name639    account_id = account_id or get_aws_account_id()640    region_name = region_name or get_region()641    if len(pattern.split("%s")) == 3:642        return pattern % (account_id, name)643    return pattern % (region_name, account_id, name)644def get_events_target_attributes(target):645    return pick_attributes(target, EVENT_TARGET_PARAMETERS)646def get_or_create_bucket(bucket_name: str, s3_client=None):647    s3_client = s3_client or connect_to_service("s3")648    try:649        return s3_client.head_bucket(Bucket=bucket_name)650    except Exception:651        return create_s3_bucket(bucket_name, s3_client=s3_client)652def create_s3_bucket(bucket_name: str, s3_client=None):653    """Creates a bucket in the region that is associated with the current request654    context, or with the given boto3 S3 client, if specified."""655    s3_client = s3_client or connect_to_service("s3")656    region = s3_client.meta.region_name657    kwargs = {}658    if region != AWS_REGION_US_EAST_1:659        kwargs = {"CreateBucketConfiguration": {"LocationConstraint": region}}660    return s3_client.create_bucket(Bucket=bucket_name, **kwargs)661def create_sqs_queue(queue_name, env=None):662    env = get_environment(env)663    # queue664    conn = connect_to_service("sqs", env=env)665    return conn.create_queue(QueueName=queue_name)666def sqs_queue_arn(queue_name, account_id=None, region_name=None):667    account_id = account_id or get_aws_account_id()668    region_name = region_name or get_region()669    queue_name = queue_name.split("/")[-1]670    return "arn:aws:sqs:%s:%s:%s" % (region_name, account_id, queue_name)671def apigateway_restapi_arn(api_id, account_id=None, region_name=None):672    account_id = account_id or get_aws_account_id()673    region_name = region_name or get_region()674    return "arn:aws:apigateway:%s:%s:/restapis/%s" % (region_name, account_id, api_id)675def sqs_queue_name(queue_arn):676    if ":" in queue_arn:677        return parse_arn(queue_arn)["resource"]678    else:679        return queue_arn680def sns_topic_arn(topic_name, account_id=None):681    account_id = account_id or get_aws_account_id()682    return "arn:aws:sns:%s:%s:%s" % (get_region(), account_id, topic_name)683def sqs_receive_message(queue_arn):684    region_name = extract_region_from_arn(queue_arn)685    client = connect_to_service("sqs", region_name=region_name)686    queue_url = get_sqs_queue_url(queue_arn)687    response = client.receive_message(QueueUrl=queue_url)688    return response689def firehose_name(firehose_arn):690    return firehose_arn.split("/")[-1]691def kinesis_stream_name(kinesis_arn):692    return kinesis_arn.split(":stream/")[-1]693def mock_aws_request_headers(694    service="dynamodb", region_name=None, access_key=None, internal=False695) -> Dict[str, str]:696    ctype = APPLICATION_AMZ_JSON_1_0697    if service == "kinesis":698        ctype = APPLICATION_AMZ_JSON_1_1699    elif service in ["sns", "sqs", "sts", "cloudformation"]:700        ctype = APPLICATION_X_WWW_FORM_URLENCODED701    # For S3 presigned URLs, we require that the client and server use the same702    # access key ID to sign requests. So try to use the access key ID for the703    # current request if available704    access_key = access_key or get_ctx_aws_access_key_id() or TEST_AWS_ACCESS_KEY_ID705    region_name = region_name or get_region()706    headers = {707        "Content-Type": ctype,708        "Accept-Encoding": "identity",709        "X-Amz-Date": "20160623T103251Z",710        "Authorization": (711            "AWS4-HMAC-SHA256 "712            + f"Credential={access_key}/20160623/{region_name}/{service}/aws4_request, "713            + "SignedHeaders=content-type;host;x-amz-date;x-amz-target, Signature=1234"714        ),715    }716    if internal:717        headers[HEADER_LOCALSTACK_ACCOUNT_ID] = get_aws_account_id()718    return headers719def inject_region_into_auth_headers(region, headers):720    auth_header = headers.get("Authorization")721    if auth_header:722        regex = r"Credential=([^/]+)/([^/]+)/([^/]+)/"723        auth_header = re.sub(regex, r"Credential=\1/\2/%s/" % region, auth_header)724        headers["Authorization"] = auth_header725def dynamodb_get_item_raw(request):726    headers = mock_aws_request_headers()727    headers["X-Amz-Target"] = "DynamoDB_20120810.GetItem"728    new_item = make_http_request(729        url=config.service_url("dynamodb"),730        method="POST",731        data=json.dumps(request),732        headers=headers,733    )734    new_item = new_item.text735    new_item = new_item and json.loads(new_item)736    return new_item737def create_dynamodb_table(738    table_name: str,739    partition_key: str,740    stream_view_type: str = None,741    region_name: str = None,742    client=None,743    wait_for_active: bool = True,744):745    """Utility method to create a DynamoDB table"""746    dynamodb = client or connect_to_service("dynamodb", region_name=region_name)747    stream_spec = {"StreamEnabled": False}748    key_schema = [{"AttributeName": partition_key, "KeyType": "HASH"}]749    attr_defs = [{"AttributeName": partition_key, "AttributeType": "S"}]750    if stream_view_type is not None:751        stream_spec = {"StreamEnabled": True, "StreamViewType": stream_view_type}752    table = None753    try:754        table = dynamodb.create_table(755            TableName=table_name,756            KeySchema=key_schema,757            AttributeDefinitions=attr_defs,758            BillingMode="PAY_PER_REQUEST",759            StreamSpecification=stream_spec,760        )761    except Exception as e:762        if "ResourceInUseException" in str(e):763            # Table already exists -> return table reference764            return connect_to_resource("dynamodb", region_name=region_name).Table(table_name)765        if "AccessDeniedException" in str(e):766            raise767    def _is_active():768        return dynamodb.describe_table(TableName=table_name)["Table"]["TableStatus"] == "ACTIVE"769    if wait_for_active:770        poll_condition(_is_active)771    return table772def get_apigateway_integration(api_id, method, path, env=None):773    apigateway = connect_to_service(service_name="apigateway", client=True, env=env)774    resources = apigateway.get_resources(restApiId=api_id, limit=100)775    resource_id = None776    for r in resources["items"]:777        if r["path"] == path:778            resource_id = r["id"]779    if not resource_id:780        raise Exception('Unable to find apigateway integration for path "%s"' % path)781    integration = apigateway.get_integration(782        restApiId=api_id, resourceId=resource_id, httpMethod=method783    )784    return integration785def get_apigateway_resource_for_path(api_id, path, parent=None, resources=None):786    if resources is None:787        apigateway = connect_to_service(service_name="apigateway")788        resources = apigateway.get_resources(restApiId=api_id, limit=100)789    if not isinstance(path, list):790        path = path.split("/")791    if not path:792        return parent793    for resource in resources:794        if resource["pathPart"] == path[0] and (not parent or parent["id"] == resource["parentId"]):795            return get_apigateway_resource_for_path(796                api_id, path[1:], parent=resource, resources=resources797            )798    return None799def get_apigateway_path_for_resource(800    api_id, resource_id, path_suffix="", resources=None, region_name=None801):802    if resources is None:803        apigateway = connect_to_service(service_name="apigateway", region_name=region_name)804        resources = apigateway.get_resources(restApiId=api_id, limit=100)["items"]805    target_resource = list(filter(lambda res: res["id"] == resource_id, resources))[0]806    path_part = target_resource.get("pathPart", "")807    if path_suffix:808        if path_part:809            path_suffix = "%s/%s" % (path_part, path_suffix)810    else:811        path_suffix = path_part812    parent_id = target_resource.get("parentId")813    if not parent_id:814        return "/%s" % path_suffix815    return get_apigateway_path_for_resource(816        api_id,817        parent_id,818        path_suffix=path_suffix,819        resources=resources,820        region_name=region_name,821    )822def create_api_gateway(823    name,824    description=None,825    resources=None,826    stage_name=None,827    enabled_api_keys=None,828    env=None,829    usage_plan_name=None,830    region_name=None,831    auth_creator_func=None,  # function that receives an api_id and returns an authorizer_id832    client=None,833):834    if enabled_api_keys is None:835        enabled_api_keys = []836    if not client:837        client = connect_to_service("apigateway", env=env, region_name=region_name)838    resources = resources or []839    stage_name = stage_name or "testing"840    usage_plan_name = usage_plan_name or "Basic Usage"841    description = description or 'Test description for API "%s"' % name842    LOG.info('Creating API resources under API Gateway "%s".', name)843    api = client.create_rest_api(name=name, description=description)844    api_id = api["id"]845    auth_id = None846    if auth_creator_func:847        auth_id = auth_creator_func(api_id)848    resources_list = client.get_resources(restApiId=api_id)849    root_res_id = resources_list["items"][0]["id"]850    # add API resources and methods851    for path, methods in resources.items():852        # create resources recursively853        parent_id = root_res_id854        for path_part in path.split("/"):855            api_resource = client.create_resource(856                restApiId=api_id, parentId=parent_id, pathPart=path_part857            )858            parent_id = api_resource["id"]859        # add methods to the API resource860        for method in methods:861            kwargs = {"authorizerId": auth_id} if auth_id else {}862            client.put_method(863                restApiId=api_id,864                resourceId=api_resource["id"],865                httpMethod=method["httpMethod"],866                authorizationType=method.get("authorizationType") or "NONE",867                apiKeyRequired=method.get("apiKeyRequired") or False,868                requestParameters=method.get("requestParameters") or {},869                requestModels=method.get("requestModels") or {},870                **kwargs,871            )872            # create integrations for this API resource/method873            integrations = method["integrations"]874            create_api_gateway_integrations(875                api_id,876                api_resource["id"],877                method,878                integrations,879                env=env,880                region_name=region_name,881                client=client,882            )883    # deploy the API gateway884    client.create_deployment(restApiId=api_id, stageName=stage_name)885    return api886def create_api_gateway_integrations(887    api_id, resource_id, method, integrations=None, env=None, region_name=None, client=None888):889    if integrations is None:890        integrations = []891    if not client:892        client = connect_to_service("apigateway", env=env, region_name=region_name)893    for integration in integrations:894        req_templates = integration.get("requestTemplates") or {}895        res_templates = integration.get("responseTemplates") or {}896        success_code = integration.get("successCode") or "200"897        client_error_code = integration.get("clientErrorCode") or "400"898        server_error_code = integration.get("serverErrorCode") or "500"899        request_parameters = integration.get("requestParameters") or {}900        # create integration901        client.put_integration(902            restApiId=api_id,903            resourceId=resource_id,904            httpMethod=method["httpMethod"],905            integrationHttpMethod=method.get("integrationHttpMethod") or method["httpMethod"],906            type=integration["type"],907            uri=integration["uri"],908            requestTemplates=req_templates,909            requestParameters=request_parameters,910        )911        response_configs = [912            {"pattern": "^2.*", "code": success_code, "res_templates": res_templates},913            {"pattern": "^4.*", "code": client_error_code, "res_templates": {}},914            {"pattern": "^5.*", "code": server_error_code, "res_templates": {}},915        ]916        # create response configs917        for response_config in response_configs:918            # create integration response919            client.put_integration_response(920                restApiId=api_id,921                resourceId=resource_id,922                httpMethod=method["httpMethod"],923                statusCode=response_config["code"],924                responseTemplates=response_config["res_templates"],925                selectionPattern=response_config["pattern"],926            )927            # create method response928            client.put_method_response(929                restApiId=api_id,930                resourceId=resource_id,931                httpMethod=method["httpMethod"],932                statusCode=response_config["code"],933            )934def apigateway_invocations_arn(lambda_uri, region_name: str = None):935    return "arn:aws:apigateway:%s:lambda:path/2015-03-31/functions/%s/invocations" % (936        region_name or get_region(),937        lambda_uri,938    )939def get_opensearch_endpoint(domain_arn: str) -> str:940    """941    Get an OpenSearch cluster endpoint by describing the cluster associated with the domain_arn942    :param domain_arn: ARN of the cluster.943    :returns: cluster endpoint944    :raises: ValueError if the domain_arn is malformed945    """946    region_name = extract_region_from_arn(domain_arn)947    if region_name is None:948        raise ValueError("unable to parse region from opensearch domain ARN")949    opensearch_client = connect_to_service(service_name="opensearch", region_name=region_name)950    domain_name = domain_arn.rpartition("/")[2]951    info = opensearch_client.describe_domain(DomainName=domain_name)952    base_domain = info["DomainStatus"]["Endpoint"]953    endpoint = base_domain if base_domain.startswith("http") else f"https://{base_domain}"954    return endpoint955def get_search_db_connection(endpoint: str, region_name: str):956    """957    Get a connection to an ElasticSearch or OpenSearch DB958    :param endpoint: cluster endpoint959    :param region_name: cluster region e.g. us-east-1960    """961    from opensearchpy import OpenSearch, RequestsHttpConnection962    from requests_aws4auth import AWS4Auth963    verify_certs = False964    use_ssl = False965    # use ssl?966    if "https://" in endpoint:967        use_ssl = True968        # TODO remove this condition once ssl certs are available for .es.localhost.localstack.cloud domains969        endpoint_netloc = urlparse(endpoint).netloc970        if not re.match(r"^.*(localhost(\.localstack\.cloud)?)(:\d+)?$", endpoint_netloc):971            verify_certs = True972    LOG.debug("Creating ES client with endpoint %s", endpoint)973    if ENV_ACCESS_KEY in os.environ and ENV_SECRET_KEY in os.environ:974        access_key = os.environ.get(ENV_ACCESS_KEY)975        secret_key = os.environ.get(ENV_SECRET_KEY)976        session_token = os.environ.get(ENV_SESSION_TOKEN)977        awsauth = AWS4Auth(access_key, secret_key, region_name, "es", session_token=session_token)978        connection_class = RequestsHttpConnection979        return OpenSearch(980            hosts=[endpoint],981            verify_certs=verify_certs,982            use_ssl=use_ssl,983            connection_class=connection_class,984            http_auth=awsauth,985        )986    return OpenSearch(hosts=[endpoint], verify_certs=verify_certs, use_ssl=use_ssl)987def create_kinesis_stream(stream_name, shards=1, env=None, delete=False):988    env = get_environment(env)989    stream = KinesisStream(id=stream_name, num_shards=shards)990    conn = connect_to_service("kinesis", env=env)991    stream.connect(conn)992    if delete:993        run_safe(lambda: stream.destroy(), print_error=False)994    stream.create()995    # Note: Returning the stream without awaiting its creation (via wait_for()) to avoid API call timeouts/retries.996    return stream997def kinesis_get_latest_records(stream_name, shard_id, count=10, env=None):998    kinesis = connect_to_service("kinesis", env=env)999    result = []1000    response = kinesis.get_shard_iterator(1001        StreamName=stream_name, ShardId=shard_id, ShardIteratorType="TRIM_HORIZON"1002    )1003    shard_iterator = response["ShardIterator"]1004    while shard_iterator:1005        records_response = kinesis.get_records(ShardIterator=shard_iterator)1006        records = records_response["Records"]1007        for record in records:1008            try:1009                record["Data"] = to_str(record["Data"])1010            except Exception:1011                pass1012        result.extend(records)1013        shard_iterator = records_response["NextShardIterator"] if records else False1014        while len(result) > count:1015            result.pop(0)1016    return result1017def get_stack_details(stack_name, region_name=None):1018    cloudformation = connect_to_service("cloudformation", region_name=region_name)1019    stacks = cloudformation.describe_stacks(StackName=stack_name)1020    for stack in stacks["Stacks"]:1021        if stack["StackName"] == stack_name:1022            return stack1023def await_stack_status(stack_name, expected_statuses, retries=20, sleep=2, region_name=None):1024    def check_stack():1025        stack = get_stack_details(stack_name, region_name=region_name)1026        if stack["StackStatus"] not in expected_statuses:1027            raise Exception(1028                'Status "%s" for stack "%s" not in expected list: %s'1029                % (stack["StackStatus"], stack_name, expected_statuses)1030            )1031        return stack1032    expected_statuses = (1033        expected_statuses if isinstance(expected_statuses, list) else [expected_statuses]1034    )1035    return retry(check_stack, retries, sleep)1036def await_stack_completion(stack_name, retries=20, sleep=2, statuses=None, region_name=None):1037    statuses = statuses or ["CREATE_COMPLETE", "UPDATE_COMPLETE", "DELETE_COMPLETE"]1038    return await_stack_status(1039        stack_name, statuses, retries=retries, sleep=sleep, region_name=region_name1040    )1041def get_ecr_repository_arn(name, account_id=None, region_name=None):1042    pattern = "arn:aws:ecr:%s:%s:repository/%s"1043    return _resource_arn(name, pattern, account_id=account_id, region_name=region_name)1044def get_route53_resolver_firewall_rule_group_arn(1045    id: str, account_id: str = None, region_name: str = None1046):1047    pattern = "arn:aws:route53resolver:%s:%s:firewall-rule-group/%s"1048    return _resource_arn(id, pattern, account_id=account_id, region_name=region_name)1049def get_route53_resolver_firewall_domain_list_arn(1050    id: str, account_id: str = None, region_name: str = None1051):1052    pattern = "arn:aws:route53resolver:%s:%s:firewall-domain-list/%s"1053    return _resource_arn(id, pattern, account_id=account_id, region_name=region_name)1054def get_route53_resolver_firewall_rule_group_associations_arn(1055    id: str, account_id: str = None, region_name: str = None1056):1057    pattern = "arn:aws:route53resolver:%s:%s:firewall-rule-group-association/%s"1058    return _resource_arn(id, pattern, account_id=account_id, region_name=region_name)1059def get_trace_id():...

rds_aurora.py

Source:rds_aurora.py

1import boto32class DbClient:3    _service = 'rds-data'4    _region = 'us-east-1'5    _database = 'voxxdb'6    _aws_access_key_id = 'AKIAVI5BSANHXPPLC3NI'7    _aws_secret_access_key = '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'8    _resource_arn = 'arn:aws:rds:us-east-1:362726687567:cluster:voxx-db-1'9    _secret_arn = 'arn:aws:secretsmanager:us-east-1:362726687567:secret:voxx-user-bYZCyV'10    def __init__(self):11        self._client = boto3.client(12            self._service,13            region_name=self._region,14            aws_access_key_id=self._aws_access_key_id,15            aws_secret_access_key=self._aws_secret_access_key,16        )17    def execute_qry(self, qry, params=[]):18        return self._client.execute_statement(19            database=self._database,20            continueAfterTimeout=False,21            includeResultMetadata=True,22            resourceArn=self._resource_arn,23            secretArn=self._secret_arn,24            parameters=params,25            sql=qry,...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.