make key in Kafka event to be optional

[Weugene]

Expected Behaviour

I can't use KafkaBaseEventModel(**event) if a record doesn't contain key field!

from aws_lambda_powertools.utilities.parser.models import KafkaBaseEventModel

event = {
    "eventSource": "SelfManagedKafka", 
    "bootstrapServers": "someBootstrapServers", 
    "records": {
        "topicName-0": [
            {
                "topic": "topicName", 
                "partition": 0, 
                "offset": 28491, 
                "timestamp": 1744827210020, 
                "key": b"",
                "timestampType": "CREATE_TIME", 
                "value": "c29tZTY0ZW5jb2RlZFZhbHVl",
                "headers": [
                    {"clientId": [64, 115, 97, 45, 112, 105, 112]}, 
                ]
            }
        ]
    }
}

KafkaBaseEventModel(**event)

it should parse the event

Current Behaviour

it writes:

Traceback (most recent call last):
  File "C:\Program Files\JetBrains\PyCharm 2022.3.2\plugins\python-ce\helpers\pydev\pydevconsole.py", line 364, in runcode
    coro = func()
           ^^^^^^
  File "<input>", line 24, in <module>
  File "C:\Users\esharaborin\repos\sa-pipeline-worker\venv\Lib\site-packages\pydantic\main.py", line 253, in __init__
    validated_self = self.__pydantic_validator__.validate_python(data, self_instance=self)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
pydantic_core._pydantic_core.ValidationError: 1 validation error for KafkaBaseEventModel
records.topicName-0.0.value
  Value error, base64 decode failed [type=value_error, input_value='some64encodedValue', input_type=str]
    For further information visit https://errors.pydantic.dev/2.11/v/value_error

Code snippet

from aws_lambda_powertools.utilities.parser.models import KafkaBaseEventModel

event = {
    "eventSource": "SelfManagedKafka", 
    "bootstrapServers": "someBootstrapServers", 
    "records": {
        "topicName-0": [
            {
                "topic": "topicName", 
                "partition": 0, 
                "offset": 28491, 
                "timestamp": 1744827210020, 
                "key": b"",
                "timestampType": "CREATE_TIME", 
                "value": "c29tZTY0ZW5jb2RlZFZhbHVl",
                "headers": [
                    {"clientId": [64, 115, 97, 45, 112, 105, 112]}, 
                ]
            }
        ]
    }
}

KafkaBaseEventModel(**event)

Possible Solution

make key optional

Steps to Reproduce

I added MWE

Powertools for AWS Lambda (Python) version

latest

AWS Lambda function runtime

3.11

Packaging format used

PyPi

Debugging logs

[boring-cyborg]

Thanks for opening your first issue here! We'll come back to you as soon as we can.
In the meantime, check out the #python channel on our Powertools for AWS Lambda Discord: Invite link

[leandrodamascena]

Hi @Weugene! Thanks for opening this issue! I thought the key was a required field in Kafka to associate this message with a partition, but reading the Confluent documentation again, I see that if the key is not present, a round-robin algorithm will be used to determine the partition. Cool, I learned something new today!

Do you want to send a PullRequest to make this field optional? I think you will have to change the field_validator as well to only run base64_decode when it is different from None.

Please let me know, if you can't fix this, I'm happy to send a PR to fix.

[tirekh]

Dear @leandrodamascena,
let me try by myself, I will create a Pull Request and then add you as a reviewer.

[leandrodamascena]

Dear @leandrodamascena,
let me try by myself, I will create a Pull Request and then add you as a reviewer.

Hey @tirekh! Sure, please let me know if you need any help! 😃