jsii
jsii copied to clipboard
aws
AttributeError: type object '<object_name>' has no attribute '__jsii_type__' when adding aws_sns LambdaSubscription
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 '<object_name>' has no attribute 'jsii_type'
Reproduction Steps
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.
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
Hey @rgibbard,
Can you share exactly what are you passing into filter_policy? thanks
@peterwoodworth an instance of S3EventBusFilterPolicy. It implements Mapping so according to the type declaration for LambdaSubscription's filter_policy argument it should work.
copy_file_filter = S3EventBusFilterPolicy()
file_copier = S3EventBusObjectCopier(
self,
"TestFileCopier",
topic=some_topic,
filter_policy=copy_file_filter,
target_bucket=some_bucket,
target_prefix="foo"
)
I'll create a simple example here for what you need to input:
policy1 = sns.SubscriptionFilter()
policy2 = sns.SubscriptionFilter()
topic.add_subscription(subs.LambdaSubscription(fn,
filter_policy={
"policy1": policy1,
"policy2": policy2
}
))
S3EventBusFilterPolicy() should return something in the format of
{
"string": SubscriptionFilter
"string": SubscriptionFilter
...
}
@peterwoodworth yeah that works, but the issue here is really that the type hint for subs.LambdaSubscription is wrong. It cannot simply take an object that isinstance(obj, typing.Mapping). It seems to only work if a built-in dict with type Dict[str, SubscriptionFilter] is passed like in your example.
.env/lib/python3.8/site-packages/aws_cdk/aws_sns_subscriptions/init.py
@jsii.implements(aws_cdk.aws_sns.ITopicSubscription)
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:
I want to pass an object that implements typing.Mapping containing Dict[str, SubscriptionFilter].
I want to pass an object that implements typing.Mapping containing Dict[str, SubscriptionFilter]
I'm not sure why you would want to do that here. Is there a reason you want to do this instead of passing in a dictionary?
When looking at our docs, it tells us it wants [Mapping[str, SubscriptionFilter]] My IDE tells me the same thing filter_policy: Mapping[str, SubscriptionFilter]
What exactly is the type hint you're seeing that's causing the confusion? And also where are you seeing it?
@peterwoodworth I want to be able to pass a fully typed object to LambdaSubscription that wraps the SubscriptionFilter objects, rather than a dictionary of magic string keys and SubscriptionFilter object values. If says it takes [Mapping[str, SubscriptionFilter]] then the S3EventBusFilterPolicy object in my example should satisfy that requirement.
As it stands, the type hint for filter_policy should probably be typing.Optional[typing.Dict[builtins.str, aws_cdk.aws_sns.SubscriptionFilter]] and not typing.Optional[typing.Mapping[builtins.str, aws_cdk.aws_sns.SubscriptionFilter]] because if you pass it anything other than a Python built-in dictionary it throws the error in the description.
Edit: In the original example, I left out some additional fields that are present in S3EventBusFilterPolicy so the object seems a bit useless, but it works as an example.