AttributeError: type object '<object_name>' has no attribute '__jsii_type__' when adding aws_sns LambdaSubscription
- Dominant language
- TypeScript
- Stars
- 2.9k
- Forks
- 267
- Avg merge
- 1d 25m
- Merged PRs (30d)
- 14
Description
### What is the problem?
Calling `sns.Topic.add_subscription(sns_subscriptions.LambdaSubscription(lambda_, filter_policy=filter_policy))` and passing a filter_policy object that implements `Mapping` throws: AttributeError: type object '' has no attribute '__jsii_type__'
### Reproduction Steps
```python
from typing import Optional, cast, Union
from collections.abc import Mapping
from aws_cdk import (
aws_s3 as s3,
aws_iam as iam,
aws_lambda as lambda_,
aws_sns as sns,
aws_sns_subscriptions as sns_subscriptions,
core
)
class S3EventBusFilterPolicy(Mapping):
default_s3_event_types = ["ObjectCreated:Put", "ObjectCreated:Copy", "ObjectCreated:CompleteMultipartUpload"]
def __init__(
self,
s3_event_type_filter: Optional[sns.SubscriptionFilter] = None
):
if s3_event_type_filter:
self._s3_event_type_filter = s3_event_type_filter
else:
self._s3_event_type_filter = sns.SubscriptionFilter.string_filter(whitelist=self.default_s3_event_types)
self._event_sub_prop_mapping = {
"eventName": self._s3_event_type_filter,
}
def __getitem__(self, key):
if key in self._event_sub_prop_mapping:
return self._event_sub_prop_mapping[key]
return KeyError(key)
def __iter__(self):
return iter(self._event_sub_prop_mapping)
def __len__(self):
return len(self._event_sub_prop_mapping)
class S3EventBusObjectCopier(core.Construct):
def __init__(
self,
scope: core.Construct,
id: str,
*,
topic: sns.Topic,
filter_policy: S3EventBusFilterPolicy,
target_bucket: s3.IBucket,
target_prefix: str
):
super().__init__(scope, id)
role = self._role(target_bucket)
self.lambda_fn = self._lambda_fn(role=role, target_bucket=target_bucket, target_prefix=target_prefix)
self._add_subscription(topic=topic, filter_policy=filter_policy)
def _role(self, bucket: s3.IBucket) -> iam.Role:
role = iam.Role(
self,
"LambdaRole",
managed_policies=[
iam.ManagedPolicy.from_aws_managed_policy_name("service-role/AWSLambdaBasicExecutionRole")
],
assumed_by=iam.ServicePrincipal("lambda.amazonaws.com")
)
bucket.grant_read_write(role)
return role
def _lambda_fn(self, role, target_bucket, target_prefix, log_level="INFO"):
return (
lambda_.Function(
self,
"LambdaFn",
description="Copy S3 object to target",
runtime=cast(lambda_.Runtime, lambda_.Runtime.PYTHON_3_8),
handler="lambda-handler.lambda_handler",
timeout=core.Duration.seconds(900),
environment={
"TARGET_BUCKET": target_bucket.bucket_name,
"TARGET_PREFIX": target_prefix,
"LOG_LEVEL": log_level
},
memory_size=512,
role=role,
code=lambda_.Code.asset("./lambda-assets/s3-event-object-copier"),
)
)
def _add_subscription(self, topic: sns.Topic, filter_policy: Union[S3EventBusFilterPolicy, Mapping]):
return topic.add_subscription(sns_subscriptions.LambdaSubscription(self.lambda_fn, filter_policy=filter_policy))
```
### What did you expect to happen?
`LambdaSusscription` has a constructer argument type hint for `filter_policy` indicating it takes a `type.Mapping` but it seems to throw this error if you pass it anything other than a Python built-in dictionary.
```python
class LambdaSubscription(
metaclass=jsii.JSIIMeta,
jsii_type="@aws-cdk/aws-sns-subscriptions.LambdaSubscription",
):
'''Use a Lambda function as a subscription target.'''
def __init__(
self,
fn: aws_cdk.aws_lambda.IFunction,
*,
dead_letter_queue: typing.Optional[aws_cdk.aws_sqs.IQueue] = None,
filter_policy: typing.Optional[typing.Mapping[builtins.str, aws_cdk.aws_sns.SubscriptionFilter]] = None,
) -> None:
```
### What actually happened?
File "/.env/lib/python3.8/site-packages/jsii/_kernel/__init__.py", line 292, in create
fqn=klass.__jsii_type__ or "Object",
AttributeError: type object 'S3EventBusFilterPolicy' has no attribute '__jsii_type__'
Subprocess exited with error 1
### CDK CLI Version
1.124.0 (build 65761fe)
### Framework Version
_No response_
### Node.js Version
7.19.1
### OS
MacOS 11.6
### Language
Python
### Language Version
3.8.9
### Other information
_No response_
Contributor guide
Research direction
Start by reproducing the failure described in the issue with a custom Mapping and a built-in dictionary, then trace jsii's kernel create path where __jsii_type__ is accessed. The fix is done when a Mapping-compatible filter_policy is accepted without requiring a __jsii_type__ attribute and the existing LambdaSubscription behavior remains intact.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python, typescript
- Domain
- developer-experience, tooling
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 42/100