(aws_events_targets): Unable to instantiate `targets.CloudWatchLogGroup` in Python due to bad typeguard checks

[anentropic]

Describe the bug

the docstring for LogGroupTargetInput shows an example:

        rule.add_target(targets.CloudWatchLogGroup(log_group,
            log_event=targets.LogGroupTargetInput.from_object_v2(
                timestamp=events.EventField.from_path("$.time"),
                message=events.EventField.from_path("$.detail-type")
            )
        ))

When attempting to use idiom this I get the following error at deploy time:

File "/home/circleci/project/infra/constructs/notifications.py", line 107, in init
self.log_target = events_targets.CloudWatchLogGroup(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/circleci/project/.venv/lib/python3.11/site-packages/jsii/_runtime.py", line 118, in call
inst = super(JSIIMeta, cast(JSIIMeta, cls)).call(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/circleci/project/.venv/lib/python3.11/site-packages/aws_cdk/aws_events_targets/init.py", line 1801, in init
props = LogGroupProps(
^^^^^^^^^^^^^^
File "/home/circleci/project/.venv/lib/python3.11/site-packages/aws_cdk/aws_events_targets/init.py", line 5798, in init
check_type(argname="argument log_event", value=log_event, expected_type=type_hints["log_event"])
File "/home/circleci/project/.venv/lib/python3.11/site-packages/aws_cdk/aws_events_targets/init.py", line 819, in check_type
return typeguard.check_type(argname=argname, value=value, expected_type=expected_type) # type:ignore
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/circleci/project/.venv/lib/python3.11/site-packages/typeguard/init.py", line 757, in check_type
checker_func(argname, value, expected_type, memo)
File "/home/circleci/project/.venv/lib/python3.11/site-packages/typeguard/init.py", line 558, in check_union
raise TypeError('type of {} must be one of ({}); got {} instead'.
TypeError: type of argument log_event must be one of (aws_cdk.aws_events_targets.LogGroupTargetInput, NoneType); got aws_cdk.aws_events._RuleTargetInputProxy instead

Regression Issue

• Select this option if this issue appears to be a regression.

Last Known Working CDK Library Version

No response

Expected Behavior

it works - my code matches the docstring example in passing log_event kwarg as log_event=targets.LogGroupTargetInput.from_object_v2(...)

Current Behavior

the error implies that the runtime return type of LogGroupTargetInput.from_object_v2 is actually aws_cdk.aws_events._RuleTargetInputProxy rather than the LogGroupTargetInput it's annotated as (and was given as the expectation to typeguard)

Reproduction Steps

I have the following code:

import json

from aws_cdk import (
    RemovalPolicy,
    Stack,
    aws_events as events,
    aws_events_targets as events_targets,
    aws_logs as logs,
)
from constructs import Construct


class MyConstruct(Construct):
    def __init__(
        self,
        scope: Construct,
        construct_id: str,
        **kwargs,
    ):
        super().__init__(scope, construct_id, **kwargs)

        message = {
            "type": "message",
            "attachments": [
                {
                    "contentType": "application/vnd.microsoft.card.adaptive",
                    "contentUrl": None,
                    "content": {
                        "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
                        "type": "AdaptiveCard",
                        "version": "1.2",
                        "body": [
                            {
                                "type": "TextBlock",
                                "text": "❗ Step Function Failure",
                            },
                            {
                                "type": "TextBlock",
                                "text": f"Time: {events.EventField.from_path('$.time')}",
                                "wrap": True,
                            },
                            {
                                "type": "TextBlock",
                                "text": f"ARN: {events.EventField.from_path('$.detail.stateMachineArn')}",
                                "wrap": True,
                            },
                            {
                                "type": "TextBlock",
                                "text": f"Error: {events.EventField.from_path('$.detail.error')}",
                                "wrap": True,
                            },
                            {
                                "type": "TextBlock",
                                "text": f"Cause: {events.EventField.from_path('$.detail.cause')}",
                                "wrap": True,
                            },
                        ],
                    },
                }
            ],
        }
        log_group = logs.LogGroup(
            self,
            "TeamsWebhookEventLogs",
            retention=logs.RetentionDays.ONE_MONTH,
            removal_policy=RemovalPolicy.DESTROY,
        )
        self.log_target = events_targets.CloudWatchLogGroup(
            log_group,
            log_event=events_targets.LogGroupTargetInput.from_object_v2(
                timestamp=events.EventField.from_path("$.time"),
                message=json.dumps(message),
            ),
        )


class MyStack(Stack):
    def __init__(
        self,
        scope: Construct,
        construct_id: str,
    ):
        super().__init__(scope, construct_id)
        MyConstruct(self, "Test")

Possible Solution

No response

Additional Information/Context

Relevant CDK code...

https://github.com/aws/aws-cdk/blob/main/packages/aws-cdk-lib/aws-events-targets/lib/log-group.ts#L54-L65

in the resulting Python CDK lib looks like:

    @jsii.member(jsii_name="fromObjectV2")
    @builtins.classmethod
    def from_object_v2(
        cls,
        *,
        message: typing.Any = None,
        timestamp: typing.Any = None,
    ) -> "LogGroupTargetInput":
        '''Pass a JSON object to the the log group event target.

        May contain strings returned by ``EventField.from()`` to substitute in parts of the
        matched event.

        :param message: The value provided here will be used in the Log "message" field. This field must be a string. If an object is passed (e.g. JSON data) it will not throw an error, but the message that makes it to CloudWatch logs will be incorrect. This is a likely scenario if doing something like: EventField.fromPath('$.detail') since in most cases the ``detail`` field contains JSON data. Default: EventField.detailType
        :param timestamp: The timestamp that will appear in the CloudWatch Logs record. Default: EventField.time
        '''
        options = LogGroupTargetInputOptions(message=message, timestamp=timestamp)

        return typing.cast("LogGroupTargetInput", jsii.sinvoke(cls, "fromObjectV2", [options]))

https://github.com/aws/aws-cdk/blob/main/packages/aws-cdk-lib/aws-events-targets/lib/log-group.ts#L76-L103

in Python CDK lib looks like:

@jsii.data_type(
    jsii_type="aws-cdk-lib.aws_events_targets.LogGroupProps",
    jsii_struct_bases=[TargetBaseProps],
    name_mapping={
        "dead_letter_queue": "deadLetterQueue",
        "max_event_age": "maxEventAge",
        "retry_attempts": "retryAttempts",
        "event": "event",
        "install_latest_aws_sdk": "installLatestAwsSdk",
        "log_event": "logEvent",
    },
)
class LogGroupProps(TargetBaseProps):
    def __init__(
        self,
        *,
        dead_letter_queue: typing.Optional["_IQueue_7ed6f679"] = None,
        max_event_age: typing.Optional["_Duration_4839e8c3"] = None,
        retry_attempts: typing.Optional[jsii.Number] = None,
        event: typing.Optional["_RuleTargetInput_6beca786"] = None,
        install_latest_aws_sdk: typing.Optional[builtins.bool] = None,
        log_event: typing.Optional["LogGroupTargetInput"] = None,
    ) -> None:
        '''Customize the CloudWatch LogGroup Event Target.

        :param dead_letter_queue: The SQS queue to be used as deadLetterQueue. Check out the `considerations for using a dead-letter queue <https://docs.aws.amazon.com/eventbridge/latest/userguide/rule-dlq.html#dlq-considerations>`_. The events not successfully delivered are automatically retried for a specified period of time, depending on the retry policy of the target. If an event is not delivered before all retry attempts are exhausted, it will be sent to the dead letter queue. Default: - no dead-letter queue
        :param max_event_age: The maximum age of a request that Lambda sends to a function for processing. Minimum value of 60. Maximum value of 86400. Default: Duration.hours(24)
        :param retry_attempts: The maximum number of times to retry when the function returns an error. Minimum value of 0. Maximum value of 185. Default: 185
        :param event: (deprecated) The event to send to the CloudWatch LogGroup. This will be the event logged into the CloudWatch LogGroup Default: - the entire EventBridge event
        :param install_latest_aws_sdk: Whether the custom resource created wll default to install latest AWS SDK. Default: - install latest AWS SDK
        :param log_event: The event to send to the CloudWatch LogGroup. This will be the event logged into the CloudWatch LogGroup Default: - the entire EventBridge event

        :exampleMetadata: infused

        Example::

            import aws_cdk.aws_logs as logs
            # log_group: logs.LogGroup
            # rule: events.Rule
            
            
            rule.add_target(targets.CloudWatchLogGroup(log_group,
                log_event=targets.LogGroupTargetInput.from_object_v2(
                    message=JSON.stringify({
                        "CustomField": "CustomValue"
                    })
                )
            ))
        '''
        if __debug__:
            type_hints = typing.get_type_hints(_typecheckingstub__955ae13452482b1f3001946153d6bfa138fb7c7e8a5c482a52178f35142a7c9c)
            check_type(argname="argument dead_letter_queue", value=dead_letter_queue, expected_type=type_hints["dead_letter_queue"])
            check_type(argname="argument max_event_age", value=max_event_age, expected_type=type_hints["max_event_age"])
            check_type(argname="argument retry_attempts", value=retry_attempts, expected_type=type_hints["retry_attempts"])
            check_type(argname="argument event", value=event, expected_type=type_hints["event"])
            check_type(argname="argument install_latest_aws_sdk", value=install_latest_aws_sdk, expected_type=type_hints["install_latest_aws_sdk"])
            check_type(argname="argument log_event", value=log_event, expected_type=type_hints["log_event"])
        self._values: typing.Dict[builtins.str, typing.Any] = {}
        if dead_letter_queue is not None:
            self._values["dead_letter_queue"] = dead_letter_queue
        if max_event_age is not None:
            self._values["max_event_age"] = max_event_age
        if retry_attempts is not None:
            self._values["retry_attempts"] = retry_attempts
        if event is not None:
            self._values["event"] = event
        if install_latest_aws_sdk is not None:
            self._values["install_latest_aws_sdk"] = install_latest_aws_sdk
        if log_event is not None:
            self._values["log_event"] = log_event

AWS CDK Library version (aws-cdk-lib)

2.235.0

AWS CDK CLI version

2.1034.0 (build 82fb0f3)

Node.js Version

v18.18.0

OS

CircleCI ubuntu-2204:2023.10.1

Language

Python

Language Version

3.11

Other information

No response

[anentropic]

wrapping it in typing.cast does not seem to help

        self.log_target = events_targets.CloudWatchLogGroup(
            log_group,
            log_event=cast(
                events_targets.LogGroupTargetInput,
                events_targets.LogGroupTargetInput.from_object_v2(
                    timestamp=events.EventField.from_path("$.time"),
                    message=json.dumps(message),
                ),
            ),
        )

File "/home/circleci/project/infra/constructs/notifications.py", line 107, in init
self.log_target = events_targets.CloudWatchLogGroup(
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/circleci/project/.venv/lib/python3.11/site-packages/jsii/_runtime.py", line 118, in call
inst = super(JSIIMeta, cast(JSIIMeta, cls)).call(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/circleci/project/.venv/lib/python3.11/site-packages/aws_cdk/aws_events_targets/init.py", line 1801, in init
props = LogGroupProps(
^^^^^^^^^^^^^^
File "/home/circleci/project/.venv/lib/python3.11/site-packages/aws_cdk/aws_events_targets/init.py", line 5798, in init
check_type(argname="argument log_event", value=log_event, expected_type=type_hints["log_event"])
File "/home/circleci/project/.venv/lib/python3.11/site-packages/aws_cdk/aws_events_targets/init.py", line 819, in check_type
return typeguard.check_type(argname=argname, value=value, expected_type=expected_type) # type:ignore
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/circleci/project/.venv/lib/python3.11/site-packages/typeguard/init.py", line 757, in check_type
checker_func(argname, value, expected_type, memo)
File "/home/circleci/project/.venv/lib/python3.11/site-packages/typeguard/init.py", line 558, in check_union
raise TypeError('type of {} must be one of ({}); got {} instead'.
TypeError: type of argument log_event must be one of (aws_cdk.aws_events_targets.LogGroupTargetInput, NoneType); got aws_cdk.aws_events._RuleTargetInputProxy instead

[anentropic]

Curiously if I look at the source of the non-v2 method it looks like more what I am seeing in the error

(I'm definitely calling from_object_v2 but maybe something is wrong under the hood?)

    @jsii.member(jsii_name="fromObject")
    @builtins.classmethod
    def from_object(
        cls,
        *,
        message: typing.Any = None,
        timestamp: typing.Any = None,
    ) -> "_RuleTargetInput_6beca786":
        '''(deprecated) Pass a JSON object to the log group event target.

        May contain strings returned by ``EventField.from()`` to substitute in parts of the
        matched event.

        :param message: The value provided here will be used in the Log "message" field. This field must be a string. If an object is passed (e.g. JSON data) it will not throw an error, but the message that makes it to CloudWatch logs will be incorrect. This is a likely scenario if doing something like: EventField.fromPath('$.detail') since in most cases the ``detail`` field contains JSON data. Default: EventField.detailType
        :param timestamp: The timestamp that will appear in the CloudWatch Logs record. Default: EventField.time

        :deprecated: use fromObjectV2

        :stability: deprecated
        '''
        options = LogGroupTargetInputOptions(message=message, timestamp=timestamp)

        return typing.cast("_RuleTargetInput_6beca786", jsii.sinvoke(cls, "fromObject", [options]))

and at the top of the file we have:

from ..aws_events import (
    IApiDestination as _IApiDestination_44cdeedd,
    IEventBus as _IEventBus_88d13111,
    IRule as _IRule_af9e3d28,
    IRuleTarget as _IRuleTarget_7a91f454,
    RuleTargetConfig as _RuleTargetConfig_4e70fe03,
    RuleTargetInput as _RuleTargetInput_6beca786,
    RuleTargetInputProperties as _RuleTargetInputProperties_38e7b0db,
)

so _RuleTargetInput_6beca786 is just RuleTargetInput from which follows this:

class _RuleTargetInputProxy(RuleTargetInput):
    @jsii.member(jsii_name="bind")
    def bind(self, rule: "_IRuleRef_4038a611") -> "RuleTargetInputProperties":
        '''Return the input properties for this input object.

        :param rule: -
        '''
        if __debug__:
            type_hints = typing.get_type_hints(_typecheckingstub__554a6befa684166f7830738df013ecfd45ccc95505b8b0c19659656cbf45fba6)
            check_type(argname="argument rule", value=rule, expected_type=type_hints["rule"])
        return typing.cast("RuleTargetInputProperties", jsii.invoke(self, "bind", [rule]))

# Adding a "__jsii_proxy_class__(): typing.Type" function to the abstract class
typing.cast(typing.Any, RuleTargetInput).__jsii_proxy_class__ = lambda : _RuleTargetInputProxy

which is the _RuleTargetInputProxy i was seeing in the my error message

[anentropic]

the original TS fromObjectV2 method:

https://github.com/aws/aws-cdk/blob/main/packages/aws-cdk-lib/aws-events-targets/lib/log-group.ts#L60-L65

  public static fromObjectV2(options?: LogGroupTargetInputOptions): LogGroupTargetInput {
    return new events.FieldAwareEventInput({
      timestamp: options?.timestamp ?? EventField.time,
      message: options?.message ?? EventField.detailType,
    }, InputType.Object);
  }

(how does this type-check in TS? ...it's annotated as LogGroupTargetInput but appears to return a FieldAwareEventInput)

the definition of that is export class FieldAwareEventInput extends RuleTargetInput

https://github.com/aws/aws-cdk/blob/main/packages/aws-cdk-lib/aws-events/lib/input.ts#L143

which seems likely why we get a _RuleTargetInputProxy in the runtime typeguard in Python ?

[anentropic]

fromObjectV2 was added in #33720

it seems that the src code is exploiting TS structural typing, presumably LogGroupTargetInput and FieldAwareEventInput are compatible shape without having any subtyping relationship ... this is a bad idea if generated code for downstream langs is going to have runtime type guards in it

it implies what is needed in the generated Python code is a Protocol, which in CDK seem to be generated by JSII for "interface" types

so it seems like there should be an ILogGroupTargetInput type def and LogGroupProps.logEvent here https://github.com/aws/aws-cdk/blob/main/packages/aws-cdk-lib/aws-events-targets/lib/log-group.ts#L94 should then expect an ILogGroupTargetInput

even then there are likely to be problems still because in CDK generated Python lib many concrete classes don't validly implement their related interface types (due to aws/jsii#4541 and reported as various issues #31454, #20278, #15150)... so you end up having to do a lot of stupid cast(iam.IPrincipal, iam.ServicePrincipal("rds.amazonaws.com")) everywhere as a Python user

...if the result of LogGroupTargetInput.from_object_v2 has to be cast as ILogGroupTargetInput to fool Pylance type checker this is unlikely to work with the runtime typeguard checks which appear to ignore typing.cast

[anentropic]

unfortunately aws-cdk-lib is pinned to an old version of typeguard lib (https://github.com/aws/jsii/blob/main/packages/%40jsii/python-runtime/setup.py#L15) so the suppress_type_checks context manager described in https://typeguard.readthedocs.io/en/stable/userguide.html#suppressing-type-checks does not exist yet

I tried:

from typeguard import typeguard_ignore

...

        self.log_target = typeguard_ignore(
            events_targets.CloudWatchLogGroup(
                log_group,
                log_event=events_targets.LogGroupTargetInput.from_object_v2(
                    timestamp=events.EventField.from_path("$.time"),
                    message=json.dumps(message),
                ),
            )
        )

but this decorator is just a wrapper over the more limited typing.no_type_check ... unclear how that even interacts with typeguard at runtime

got same error when deployed:

TypeError: type of argument log_event must be one of (aws_cdk.aws_events_targets.LogGroupTargetInput, NoneType); got aws_cdk.aws_events._RuleTargetInputProxy instead