aws_s3.Bucket.from_bucket_name : returns _BaseBucketProxy object instead of IBucket

[kamari-swami]

Describe the bug

existing_bucket_name = "test_bucket"

get_bucket = Bucket.from_bucket_name(scope, id, existing_bucket_name)

cloudformation_distro = Distribution(....., log_bucket=get_bucket,)

Error:

    test = MeteredDistribution(
  File "/.venv/lib/python3.10/site-packages/jsii/_runtime.py", line 118, in __call__
    inst = super(JSIIMeta, cast(JSIIMeta, cls)).__call__(*args, **kwargs)
  File "/.venv/lib/python3.10/site-packages/jsii/_runtime.py", line 118, in __call__
    inst = super(JSIIMeta, cast(JSIIMeta, cls)).__call__(*args, **kwargs)
  File "/.venv/lib/python3.10/site-packages/aws_cdk/aws_lambda_event_sources/__init__.py", line 1147, in __init__
    check_type(argname="argument bucket", value=bucket, expected_type=type_hints["bucket"])
  File "/.venv/lib/python3.10/site-packages/typeguard/__init__.py", line 785, in check_type
    raise TypeError(
TypeError: type of argument bucket must be aws_cdk.aws_s3.Bucket; got aws_cdk.aws_s3._BucketBaseProxy instead

Expected Behavior

Bucket.from_bucket_name should return IBucket object as per the definition of the method:

def from_bucket_name(
        cls,
        scope: _constructs_77d1e7e8.Construct,
        id: builtins.str,
        bucket_name: builtins.str,
    ) -> IBucket:

Current Behavior

Returns aws_cdk.aws_s3._BucketBaseProxy instead

Reproduction Steps

as above

Possible Solution

Not known

Additional Information/Context

No response

CDK CLI Version

2.147.1 (build d3695d4)

Framework Version

No response

Node.js Version

v20.11.0

OS

Ubuntu v22

Language

Python

Language Version

Python 3.10.12

Other information

No response

[ashishdhingra]

Hi @kamari-swami,

Good morning. Thanks for opening the issue. Could you please share minimal self reproduction CDK code to troubleshoot the issue. Using AWS CDK version 2.149.0 (build c8e5924), the below Python CDK code:

from aws_cdk import (
    Stack,
    aws_s3 as s3,
    aws_cloudfront as cloudfront,
    aws_cloudfront_origins as cloudfrontOrigins
)
from constructs import Construct

class Issue30885Stack(Stack):

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        bucket = s3.Bucket.from_bucket_name(
            self, 
            id="MyBucket",
            bucket_name="testbucket-someissue"
        )

        cloudfrontDistribution = cloudfront.Distribution(
            self,
            id="MyCloudfrontDistribution",
            default_behavior=cloudfront.BehaviorOptions(
                allowed_methods=cloudfront.AllowedMethods.ALLOW_ALL,
                origin=cloudfrontOrigins.S3Origin(
                    bucket=bucket,
                    origin_path="/"
                )
            ),
            log_bucket=bucket
        )

successfully synthesizes to the below CloudFormation template:

Resources:
  MyCloudfrontDistributionOrigin1S3OriginF5116CA8:
    Type: AWS::CloudFront::CloudFrontOriginAccessIdentity
    Properties:
      CloudFrontOriginAccessIdentityConfig:
        Comment: Identity for Issue30885StackMyCloudfrontDistributionOrigin1D7159771
    Metadata:
      aws:cdk:path: Issue30885Stack/MyCloudfrontDistribution/Origin1/S3Origin/Resource
  MyCloudfrontDistribution7D54D3CA:
    Type: AWS::CloudFront::Distribution
    Properties:
      DistributionConfig:
        DefaultCacheBehavior:
          AllowedMethods:
            - GET
            - HEAD
            - OPTIONS
            - PUT
            - PATCH
            - POST
            - DELETE
          CachePolicyId: 658327ea-f89d-4fab-a63d-7e88639e58f6
          Compress: true
          TargetOriginId: Issue30885StackMyCloudfrontDistributionOrigin1D7159771
          ViewerProtocolPolicy: allow-all
        Enabled: true
        HttpVersion: http2
        IPV6Enabled: true
        Logging:
          Bucket:
            Fn::Join:
              - ""
              - - testbucket-someissue.s3.
                - Ref: AWS::Region
                - "."
                - Ref: AWS::URLSuffix
        Origins:
          - DomainName:
              Fn::Join:
                - ""
                - - testbucket-someissue.s3.
                  - Ref: AWS::Region
                  - "."
                  - Ref: AWS::URLSuffix
            Id: Issue30885StackMyCloudfrontDistributionOrigin1D7159771
            OriginPath: ""
            S3OriginConfig:
              OriginAccessIdentity:
                Fn::Join:
                  - ""
                  - - origin-access-identity/cloudfront/
                    - Ref: MyCloudfrontDistributionOrigin1S3OriginF5116CA8
...
...

Thanks,
Ashish

[kamari-swami]

Hi @ashishdhingra Thanks for response. Here is how to reproduce:

custom-constructs.py:

class SingletonS3Bucket:
    _instance = None

    def __new__(cls, scope: Construct, construct_id: str, **kwargs):
        if not cls._instance:
            cls._instance = type("SingletonS3Bucket", (), {})     # = super().__new__(cls)
            
            existing_bucket_name = kwargs['bucket_name'] 
            existing_bucket = Bucket.from_bucket_name(scope, construct_id, existing_bucket_name)   # class <_BaseBucketProxy> 
            
            if not existing_bucket.bucket_name == existing_bucket_name:
                cls._instance.bucket = Bucket(scope, construct_id, **kwargs)
            else:
                cls._instance.bucket = existing_bucket
        return cls._instance


class MeteredDistribution(Distribution):

    def __init__(self, scope: Construct, construct_id: str, default_behavior: BehaviorOptions, **kwargs):
        singleton_bucket = SingletonS3Bucket(scope, construct_id, bucket_name='log_bucket')
        log_bucket = singleton_bucket.bucket   # class <_BaseBucketProxy> 
            super().__init__(scope, construct_id,  default_behavior=default_behavior,log_bucket=log_bucket, **kwargs)

my-stack.py:

from custom-contructs import MeteredDistribution


class PreviewStack(Stack):
    def __init__(
        self,
        scope: Construct,
        construct_id: str,
        env: Environment,
        **kwargs,
    ) -> None:
        super().__init__(scope, construct_id, env=env, **kwargs)

        test = MeteredDistribution(
                    self,
                    f"metered-distro-5",
                    default_behavior=BehaviorOptions(
                        allowed_methods=AllowedMethods.ALLOW_GET_HEAD_OPTIONS,
                        origin_request_policy=origin_request_policy,
                        viewer_protocol_policy=ViewerProtocolPolicy.REDIRECT_TO_HTTPS,
                        origin=my_origin,
                        cache_policy=CachePolicy.CACHING_DISABLED,
                    ),
                    minimum_protocol_version=SecurityPolicyProtocol.TLS_V1_2_2021,
                    certificate=certificate,
                    domain_names=[f"metered-distro-5.{previews_domain}"],
                    default_root_object="index.html",
                    enable_logging=True,
                    log_file_prefix="test",
                )

This results into :

Error:

    test = MeteredDistribution(
  File "/.venv/lib/python3.10/site-packages/jsii/_runtime.py", line 118, in __call__
    inst = super(JSIIMeta, cast(JSIIMeta, cls)).__call__(*args, **kwargs)
  File "/.venv/lib/python3.10/site-packages/jsii/_runtime.py", line 118, in __call__
    inst = super(JSIIMeta, cast(JSIIMeta, cls)).__call__(*args, **kwargs)
  File "/.venv/lib/python3.10/site-packages/aws_cdk/aws_lambda_event_sources/__init__.py", line 1147, in __init__
    check_type(argname="argument bucket", value=bucket, expected_type=type_hints["bucket"])
  File "/.venv/lib/python3.10/site-packages/typeguard/__init__.py", line 785, in check_type
    raise TypeError(
TypeError: type of argument bucket must be aws_cdk.aws_s3.Bucket; got aws_cdk.aws_s3._BucketBaseProxy instead```

[ashishdhingra]

@kamari-swami Good morning. Thanks for sharing the code snippet. While I'm not Python expert and unsure on how type system works, looking at the code below:

existing_bucket = Bucket.from_bucket_name(scope, construct_id, existing_bucket_name)   # class <_BaseBucketProxy> 

if not existing_bucket.bucket_name == existing_bucket_name:
    cls._instance.bucket = Bucket(scope, construct_id, **kwargs)
else:
    cls._instance.bucket = existing_bucket

looks like you are mixing types.

Bucket(scope, construct_id, **kwargs) constructs an object of type Bucket. Whereas, Bucket.from_bucket_name(scope, construct_id, existing_bucket_name) returns object typed to IBucket. IBucket.

It's the BucketBase class that implements IBucket looking at decompiled code:

@jsii.implements(IBucket)
class BucketBase(
    _Resource_45bc6135,
    metaclass=jsii.JSIIAbstractClass,
    jsii_type="aws-cdk-lib.aws_s3.BucketBase",
):

That too is a JSII hint. (JSII is a layer that translates CDK Python code to TypeScript code).

Could you please test what happens:

• When you explicitly assign Bucket(scope, construct_id, bucket_name) to log_bucket and use it in MeteredDistribution?
• When you explicitly assign Bucket.from_bucket_name(scope, construct_id, existing_bucket_name) to log_bucket and use it in MeteredDistribution?

Thanks,
Ashish

[github-actions]

This issue has not received a response in a while. If you want to keep this issue open, please leave a comment below and auto-close will be canceled.

[kamari-swami]

@ashishdhingra , i will test and provide results as asked.

[kamari-swami]

Hi @ashishdhingra ,

Both the scenario's below work without error:

scenario-1 (log_bucket=Bucket())

class MeteredDistribution(Distribution):

    def __init__(self, scope: Construct, construct_id: str, default_behavior: BehaviorOptions, **kwargs):
        singleton_bucket = SingletonS3Bucket(scope, construct_id, bucket_name='log_bucket')
        log_bucket = Bucket(scope, id, name)
        super().__init__(scope, construct_id,  default_behavior=default_behavior,log_bucket=log_bucket, **kwargs)

scenario-2 (log_bucket = Bucket.from_bucket_name())

class MeteredDistribution(Distribution):

    def __init__(self, scope: Construct, construct_id: str, default_behavior: BehaviorOptions, **kwargs):
        singleton_bucket = SingletonS3Bucket(scope, construct_id, bucket_name='log_bucket')
        log_bucket = Bucket.from_bucket_name(scope, id, name)
        super().__init__(scope, construct_id,  default_behavior=default_behavior,log_bucket=log_bucket, **kwargs)

[ashishdhingra]

@kamari-swami Thanks for your reply. As mentioned in #30885 (comment), you should avoid mixing types in your helper method.

Also, I guess your scenario is similar to the one mentioned in https://stackoverflow.com/questions/75294855/how-to-check-if-s3-bucket-exist-and-if-not-create-it-if-yes-take-it-in-typescri, where you are checking if the foreign bucket already exists; if not, create bucket as part of CDK stack. You can import existing resources into CDK stack or create new resources as part of CDK stack. This information should be available beforehand. Although, you could rely on AWS SDK calls to inspect existing bucket, but it might not be always scalable for scenarios where bucket was initially created as part of CDK stack and the CDK code is redeployed which checks again for existing bucket created by CDK stack.

Thanks,
Ashish

[github-actions]

This issue has not received a response in a while. If you want to keep this issue open, please leave a comment below and auto-close will be canceled.

[kamari-swami]

@ashishdhingra ,
I think your comment does not answer my original problem at hand. It does say the Bucket and .fromBucketName() returns different object types, but that is OK for me - whether I get Bucket object or IBucket object.

Cloudfront takes both :) for log_bucket.

In my code, SingletonS3Bucket constructor is holder of either Bucket() or IBucket object but not both (so there is no mixing happening :))

The other link in your response does not either gives a clean/clear solution to achieve.

Ask is:
Check if bucket exists in the stack with given name - if yes, re-use, if not, create new one. Preferably in helper function that can be re-used within multiple stacks...without re-repeating the same code again

Your suggestion worked in two scenario's that I tested after your suggestion - thanks for that,

But if I would like to re-use the logic as helper code (the one that checks if bucket exists in a stack, if yes, return IBucket, if not, return Bucket object) then i think it is not possible (as demonstrated in SingletonS3Bucket() example).

Any suggestion how can I use this if-else check as standalone helper function w/o repeating the same in every stack?

[ashishdhingra]

@ashishdhingra , I think your comment does not answer my original problem at hand. It does say the Bucket and .fromBucketName() returns different object types, but that is OK for me - whether I get Bucket object or IBucket object.

Cloudfront takes both :) for log_bucket.

In my code, SingletonS3Bucket constructor is holder of either Bucket() or IBucket object but not both (so there is no mixing happening :))

The other link in your response does not either gives a clean/clear solution to achieve.

Ask is: Check if bucket exists in the stack with given name - if yes, re-use, if not, create new one. Preferably in helper function that can be re-used within multiple stacks...without re-repeating the same code again

Your suggestion worked in two scenario's that I tested after your suggestion - thanks for that,

But if I would like to re-use the logic as helper code (the one that checks if bucket exists in a stack, if yes, return IBucket, if not, return Bucket object) then i think it is not possible (as demonstrated in SingletonS3Bucket() example).

Any suggestion how can I use this if-else check as standalone helper function w/o repeating the same in every stack?

@kamari-swami In CDK, you can import existing resources as mentioned here https://medium.com/@glasshost/import-an-existing-s3-bucket-in-aws-cdk-295e0ea787f7. The already existing bucket is not maintained by CDK. Bucket.fromBucketName() is a shortcut to import existing bucket into CDK stack. It would not make any AWS API call to hydrate bucket properties. For instance, running cdk synth on the below code:

import * as iam from 'aws-cdk-lib/aws-iam';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as cdk from 'aws-cdk-lib';

export class CdktestStack extends cdk.Stack {
  constructor(scope: cdk.App, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    const importedBucketFromName = s3.Bucket.fromBucketName(
      this,
      'imported-bucket-from-name',
      'testbucket-issue',
    );

    console.log('bucket name 👉', importedBucketFromName.bucketName);
    console.log('bucket arn 👉', importedBucketFromName.bucketArn);

    // using methods on the imported bucket
    importedBucketFromName.grantRead(new iam.AccountRootPrincipal());
  }
}

returns below output:

bucket name 👉 testbucket-issue
bucket arn 👉 arn:${Token[AWS.Partition.12]}:s3:::testbucket-issue
Resources:
...

The bucket ARN is just a token that would be resolved during deployment. Actually, the above bucket doesn't exist. So you cannot rely on Bucket.fromBucketName() for checking bucket existence.

For your scenario, the workaround is to make AWS SDK call to check if bucket already exists. Thereafter, you could pass the existence flag to CDK stack to make a decision on whether to create a new bucket or use existing one. Kindly refer https://stackoverflow.com/questions/75294855/how-to-check-if-s3-bucket-exist-and-if-not-create-it-if-yes-take-it-in-typescri for an example (not sure if it works, but worth giving a try).

Thanks,
Ashish

[github-actions]

This issue has not received a response in a while. If you want to keep this issue open, please leave a comment below and auto-close will be canceled.