Problem with cdk code to create a step function to call dynamodb to put an item

Viewed 536

CDK is great but I'm really struggling with how to deploy a step function to put an item into a dynamodb table with the python sdk. I eventually want to provide the item via event bridge.

When I get errors I'm not sure if they are real errors or not:

I've tried this:

  event_table = dynamodb.Table(self, 'events',
        partition_key=dynamodb.Attribute(name='id', type=dynamodb.AttributeType.STRING),
        billing_mode=dynamodb.BillingMode.PAY_PER_REQUEST, removal_policy=RemovalPolicy.DESTROY)

    map = {'id': {'S.$': '$.id'}}
    dynamo_put_task = CallDynamoDB.put_item(item=map, table_name=event_table.table_name)

    task = Task(self, 'invoke-notification-service', task=dynamo_put_task)
    state_machine = StateMachine(self, 'save-event', definition=task)
    Rule(self, 'rule_name', targets=[(SfnStateMachine(state_machine))], event_bus=event_bus, event_pattern=EventPattern(account=[Stack.of(self).account]))

I really don't know how to use a DynamoAttributeValueMap as this is just a shell in python which is why I'm not using one.

And even for this to run I've had to remove the following line from DynamoPutItemProps:

if isinstance(item, dict): item = DynamoAttributeValueMap(**item)

otherwise I get an "unknown keyword argument id" when I try to run this. Which kinda makes sense as it's trying to unwrap a map and pass it into a constructor that takes no parameters (just self).

A simple example of creating a step function with cdk to write to dynamodb would be very helpful.

I really don't know how to instantiate that map.

The code that I've got - in the example - at least generates the cloudformation template but the item is empty. It is just {}.

When I try to do something like this cause I think that's what I'm supposed to do...

map = DynamoAttributeValueMap()
map._values = {'id': DynamoAttributeValue().with_s('$.id')}

To try and force values into that map I get error:

TypeError: Don't know how to convert object to JSON: DynamoAttributeValueMap(id=)

So then from this, I assume that this code should work:

dynamo_put_task = CallDynamoDB.put_item(item={'id': DynamoAttributeValue().with_s('1234')},
table_name=event_table.table_name, return_values=DynamoReturnValues.ALL_NEW)

This generates this template:

 Type: AWS::StepFunctions::StateMachine
    Properties:
      DefinitionString:
        Fn::Join:
          - ""
          - - '{"StartAt":"invoke-notification-service","States":{"invoke-notification-service":{"End":true,"Parameters":{"Item":{},"TableName":"'
            - Ref: events26E65764
            - '","ReturnValues":"ALL_NEW"},"Type":"Task","Resource":"arn:'
            - Ref: AWS::Partition
            - :states:::dynamodb:putItem"}}}
      RoleArn:
        Fn::GetAtt:
          - saveeventRole24A1910F
          - Arn

Notice that the Item is blank.

Of course this is with the

if isinstance(item, dict): item = DynamoAttributeValueMap(**item)

from DynamoPutItemsProps removed.

So is there a bug? Just a plain worked example would be useful.

I do suspectand I suspect there might be a bug as nothing works unless I remove that line of code.

0 Answers
Related