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: 

まず、提供された reproducer.py スクリプトで動作を再現し、from_http エントリーポイントと cloudevents.http.event.CloudEvent の表現を調査します。解析された属性にどのようにアクセスすることが意図されているかを確認し、次に、文書化されたアクセス方法で specversion、type、source が公開されることを確認して、回帰チェックで動作を検証します。