def _get_eventhub_content()

in azurefunctions-extensions-bindings-eventhub/azurefunctions/extensions/bindings/eventhub/eventData.py [0:0]

• 7 lines of code
• 3 McCabe index (conditional complexity)

    def _get_eventhub_content(self, content):
        """
        When receiving the EventBindingData, the content field is in the form of bytes.
        This content must be decoded in order to construct an EventData object from the
        azure.eventhub SDK. The .NET worker uses the Azure.Core.Amqp library to do this:
        https://github.com/Azure/azure-functions-dotnet-worker/blob/main/extensions/Worker.Extensions.EventHubs/src/EventDataConverter.cs#L45
        """
        if content:
            try:
                return uamqp.Message().decode_from_bytes(content)
            except Exception as e:
                raise ValueError(f"Failed to decode EventHub content: {e}") from e

        return None