The `from_http` returns CloudEvent with inaccessible attributes

Expected Behavior

The from_http function should return a CloudEvent object where attributes like specversion, type, and source are accessible via event["attributes"] or similar documented methods.

Actual Behavior

But for me the from_http function returns a CloudEvent object with an empty attributes field. Attempting to access specversion, type, or source returns missing.

Steps to Reproduce the Problem

1. Save the following script as reproducer.py:

import logging
from cloudevents.http import from_http
from http.server import BaseHTTPRequestHandler, HTTPServer

logging.basicConfig(level=logging.DEBUG)

class CloudEventHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        headers = {key.lower(): value for key, value in self.headers.items()}
        content_length = int(self.headers.get("content-length", 0))
        body = self.rfile.read(content_length).decode("utf-8")

        logging.info(f"Received headers: {headers}")
        logging.info(f"Received body: {body}")

        try:
            event = from_http(headers, body)
            logging.info(f"Parsed CloudEvent: {event}")

            logging.info(f"Type of parsed event: {type(event)}")
            logging.info(f"Event attributes: {event.get('attributes', {})}")

            specversion = event.get("attributes", {}).get("specversion", "missing")
            event_type = event.get("attributes", {}).get("type", "missing")
            source = event.get("attributes", {}).get("source", "missing")

            logging.info(f"Specversion: {specversion}, Type: {event_type}, Source: {source}")

            response_body = {
                "specversion": specversion,
                "type": event_type,
                "source": source,
            }

            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(str(response_body).encode("utf-8"))

        except Exception as e:
            logging.error(f"Error parsing CloudEvent: {e}")
            self.send_response(500)
            self.end_headers()
            self.wfile.write(f"Error: {e}".encode("utf-8"))

def run(server_class=HTTPServer, handler_class=CloudEventHandler, port=8080):
    server_address = ("", port)
    httpd = server_class(server_address, handler_class)
    logging.info(f"Starting HTTP server on port {port}")
    httpd.serve_forever()

if __name__ == "__main__":
    run()

2. Run the script via python reproducer.py

3. Access the server, like:

curl -v -X POST \
    -H "Content-Type: application/json" \
    -H "ce-specversion: 1.0" \
    -H "ce-type: test.event.type" \
    -H "ce-source: /my/source" \
    -H "ce-id: 12345" \
    -d '{"key":"value"}' \
    http://127.0.0.1:8080

4. See the logs:

INFO:root:Received headers: {'host': '127.0.0.1:8080', 'user-agent': 'curl/8.9.1', 'accept': '*/*', 'content-type': 'application/json', 'ce-specversion': '1.0', 'ce-type': 'test.event.type', 'ce-source': '/my/source', 'ce-id': '12345', 'content-length': '15'}
INFO:root:Received body: {"key":"value"}
INFO:root:Parsed CloudEvent: {'attributes': {'specversion': '1.0', 'id': '12345', 'source': '/my/source', 'type': 'test.event.type', 'datacontenttype': 'application/json', 'time': '2024-12-19T09:29:20.297551+00:00'}, 'data': {'key': 'value'}}
INFO:root:Type of parsed event: <class 'cloudevents.http.event.CloudEvent'>
INFO:root:Event attributes: {}
INFO:root:Specversion: missing, Type: missing, Source: missing

Specifications

• Platform: Linux (Fedora release 41)
• Python Version: Python 3.13.0
• SDK-Version: via pip show cloudevents:

Name: cloudevents
Version: 1.11.0
Summary: CloudEvents Python SDK
Home-page: https://github.com/cloudevents/sdk-python
Author: The Cloud Events Contributors
Author-email: cncfcloudevents@gmail.com
License: https://www.apache.org/licenses/LICENSE-2.0
Location: /home/<user-name>/.local/lib/python3.13/site-packages
Requires: deprecation
Required-by: 

Beginnen Sie damit, das Verhalten mit dem bereitgestellten reproducer.py-Skript zu reproduzieren, und untersuchen Sie den from_http-Einstiegspunkt sowie die cloudevents.http.event.CloudEvent-Darstellung. Überprüfen Sie, wie auf geparste Attribute zugegriffen werden soll, stellen Sie anschließend sicher, dass die dokumentierte Zugriffsmethode specversion, type und source verfügbar macht, und bestätigen Sie das Verhalten mit einer Regressionprüfung.