⚠️ Add strongly typed EventNotifications (#1538) (#1591)

* wip generating push payload events

[skip ci]

* improve generation

[skip ci]

* generate related object method

* Return a union from parse thin event & add UnknownThinEvent

* fix unused import

* fix related_object generation and add big test

* Some name cleanup

* rename thin_event

* small fixes

* export UnknownEventNotification

* fix reason parsing

* rename thin_event

* fix example & re-export event notifications

* move some imports

* swap event fetch_related_object to stripe-context

* update docstring

* update comment

Co-authored-by: David Brownman <[email protected]>
Co-authored-by: Ramya Rao <[email protected]>

Author: ramya-stripe

examples/event_notification_webhook_handler.py

@@ -0,0 +1,55 @@
+"""
+event_notification_webhook_handler.py - receive and process thin events like the
+v1.billing.meter.error_report_triggered event.
+
+In this example, we:
+    - create a StripeClient called client
+    - use client.parse_event_notification() to parse the received notification webhook body
+    - if its type is "v1.billing.meter.error_report_triggered":
+        - call event_notification.fetch_event() to retrieve the full event object
+        - call event_notification.fetch_related_object() to retrieve the Meter that failed
+        - log info about the failure
+"""
+
+import os
+from stripe import StripeClient
+
+from flask import Flask, request, jsonify
+
+app = Flask(__name__)
+api_key = os.environ.get("STRIPE_API_KEY", "")
+webhook_secret = os.environ.get("WEBHOOK_SECRET", "")
+
+client = StripeClient(api_key)
+
+
+@app.route("/webhook", methods=["POST"])
+def webhook():
+    webhook_body = request.data
+    sig_header = request.headers.get("Stripe-Signature")
+
+    try:
+        event_notification = client.parse_event_notification(
+            webhook_body, sig_header, webhook_secret
+        )
+
+        # Fetch the event data to understand the failure
+        if (
+            event_notification.type
+            == "v1.billing.meter.error_report_triggered"
+        ):
+            meter = event_notification.fetch_related_object()
+            event = event_notification.fetch_event()
+            print(
+                f"Err! Meter {meter.id} had a problem (more info: {event.data.developer_message_summary})"
+            )
+            # Record the failures and alert your team
+            # Add your logic here
+
+        return jsonify(success=True), 200
+    except Exception as e:
+        return jsonify(error=str(e)), 400
+
+
+if __name__ == "__main__":
+    app.run(port=4242)

examples/thinevent_webhook_handler.py

@@ -35,6 +35,7 @@ class TypingImportsChecker:
         "Unpack",
         "Awaitable",
         "Never",
+        "override",
     ]
 
     allowed_typing_imports = [

flake8_stripe/flake8_stripe.py

@@ -15,6 +15,11 @@ test *args: install-test-deps
     # configured in pyproject.toml
     pytest "$@"
 
+# run a single test by name
+test-one test_name: install-test-deps
+    # don't use all cores, there's a spin up time to that and we're only using one test
+    pytest -k "{{ test_name }}" -n 0
+
 # ⭐ check for potential mistakes
 lint: install-dev-deps
     python -m flake8 --show-source stripe tests

justfile

@@ -60,6 +60,7 @@ include = [
   "tests/test_generated_examples.py",
   "tests/test_exports.py",
   "tests/test_http_client.py",
+  "tests/test_v2_event.py",
 ]
 exclude = ["build", "**/__pycache__"]
 reportMissingTypeArgument = true

pyproject.toml

@@ -100,8 +100,6 @@ def _warn_if_mismatched_proxy():
 # StripeClient
 from stripe._stripe_client import StripeClient as StripeClient  # noqa
 
-from stripe.v2._event import ThinEvent as ThinEvent  # noqa
-
 
 # Sets some basic information about the running application that's sent along
 # with API requests. Useful for plugin authors to identify their plugin when

stripe/__init__.py

@@ -861,7 +861,7 @@ async def request_raw_async(
 
         return rcontent, rcode, rheaders
 
-    def _should_handle_code_as_error(self, rcode):
+    def _should_handle_code_as_error(self, rcode: int) -> bool:
         return not 200 <= rcode < 300
 
     def _interpret_response(

stripe/_api_requestor.py

@@ -27,9 +27,10 @@
 from stripe._util import _convert_to_stripe_object, get_api_mode, deprecated  # noqa: F401
 from stripe._webhook import Webhook, WebhookSignature
 from stripe._event import Event
-from stripe.v2._event import ThinEvent
+from stripe.v2._event import EventNotification
 
 from typing import Any, Dict, Optional, Union, cast
+from typing_extensions import TYPE_CHECKING
 
 # Non-generated services
 from stripe._oauth_service import OAuthService
@@ -122,6 +123,9 @@
 from stripe._webhook_endpoint_service import WebhookEndpointService
 # services: The end of the section generated from our OpenAPI spec
 
+if TYPE_CHECKING:
+    from stripe.events._event_classes import ALL_EVENT_NOTIFICATIONS
+
 
 class StripeClient(object):
     def __init__(
@@ -201,13 +205,18 @@ def __init__(
         self.v2 = V2Services(self._requestor)
         # top-level services: The end of the section generated from our OpenAPI spec
 
-    def parse_thin_event(
+    def parse_event_notification(
         self,
         raw: Union[bytes, str, bytearray],
         sig_header: str,
         secret: str,
         tolerance: int = Webhook.DEFAULT_TOLERANCE,
-    ) -> ThinEvent:
+    ) -> "ALL_EVENT_NOTIFICATIONS":
+        """
+        This should be your main method for interacting with `EventNotifications`. It's the V2 equivalent of `construct_event()`, but with better typing support.
+
+        It returns a union representing all known `EventNotification` classes. They have a `type` property that can be used for narrowing, which will get you very specific type support. If parsing an event the SDK isn't familiar with, it'll instead return `UnknownEventNotification`. That's not reflected in the return type of the function (because it messes up type narrowing) but is otherwise intended.
+        """
         payload = (
             cast(Union[bytes, bytearray], raw).decode("utf-8")
             if hasattr(raw, "decode")
@@ -216,7 +225,10 @@ def parse_thin_event(
 
         WebhookSignature.verify_header(payload, sig_header, secret, tolerance)
 
-        return ThinEvent(payload)
+        return cast(
+            "ALL_EVENT_NOTIFICATIONS",
+            EventNotification.from_json(payload, self),
+        )
 
     def construct_event(
         self,
@@ -247,6 +259,9 @@ def raw_request(self, method_: str, url_: str, **params):
 
         stripe_context = params.pop("stripe_context", None)
 
+        # we manually pass usage in event internals, so use those if available
+        usage = params.pop("usage", ["raw_request"])
+
         # stripe-context goes *here* and not in api_requestor. Properties
         # go on api_requestor when you want them to persist onto requests
         # made when you call instance methods on APIResources that come from
@@ -263,7 +278,7 @@ def raw_request(self, method_: str, url_: str, **params):
             options=options,
             base_address=base_address,
             api_mode=api_mode,
-            usage=["raw_request"],
+            usage=usage,
         )
 
         return self._requestor._interpret_response(
@@ -297,6 +312,9 @@ def deserialize(
         *,
         api_mode: ApiMode,
     ) -> StripeObject:
+        """
+        Used to translate the result of a `raw_request` into a StripeObject.
+        """
         return _convert_to_stripe_object(
             resp=resp,
             params=params,

stripe/_stripe_client.py

@@ -192,12 +192,6 @@ def secure_compare(val1, val2):
         return result == 0
 
 
-def get_thin_event_classes():
-    from stripe.events._event_classes import THIN_EVENT_CLASSES
-
-    return THIN_EVENT_CLASSES
-
-
 def get_object_classes(api_mode):
     # This is here to avoid a circular dependency
     if api_mode == "V2":
@@ -331,8 +325,10 @@ def _convert_to_stripe_object(
 
                 klass = DeletedObject
             elif api_mode == "V2" and klass_name == "v2.core.event":
+                from stripe.events._event_classes import V2_EVENT_CLASS_LOOKUP
+
                 event_name = resp.get("type", "")
-                klass = get_thin_event_classes().get(
+                klass = V2_EVENT_CLASS_LOOKUP.get(
                     event_name, stripe.StripeObject
                 )
             else:
@@ -426,9 +422,11 @@ def sanitize_id(id):
     return quotedId
 
 
-def get_api_mode(url):
+def get_api_mode(url: str) -> ApiMode:
     if url.startswith("/v2"):
         return "V2"
+
+    # if urls aren't explicitly marked as v1, they're assumed to be v1
     else:
         return "V1"