⚠️ Add strongly typed EventNotifications

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)

flake8_stripe/flake8_stripe.py

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

justfile

@@ -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

pyproject.toml

@@ -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

stripe/__init__.py

@@ -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/_api_requestor.py

@@ -827,7 +827,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/_stripe_client.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
@@ -113,6 +114,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__(
@@ -192,13 +196,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")
@@ -207,7 +216,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,
@@ -238,6 +250,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
@@ -254,7 +269,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(
@@ -288,6 +303,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/_util.py

@@ -9,7 +9,7 @@
 
 from stripe._api_mode import ApiMode
 
-from urllib.parse import parse_qsl, quote_plus  # noqa: F401
+from urllib.parse import parse_qsl, quote_plus, urlparse  # noqa: F401
 
 from typing_extensions import Type, TYPE_CHECKING
 from typing import (
@@ -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"
 

stripe/events/__init__.py

@@ -1,11 +1,17 @@
 # -*- coding: utf-8 -*-
 # File generated from our OpenAPI spec
+from stripe.events._event_classes import (
+    ALL_EVENT_NOTIFICATIONS as ALL_EVENT_NOTIFICATIONS,
+)
 from stripe.events._v1_billing_meter_error_report_triggered_event import (
     V1BillingMeterErrorReportTriggeredEvent as V1BillingMeterErrorReportTriggeredEvent,
+    V1BillingMeterErrorReportTriggeredEventNotification as V1BillingMeterErrorReportTriggeredEventNotification,
 )
 from stripe.events._v1_billing_meter_no_meter_found_event import (
     V1BillingMeterNoMeterFoundEvent as V1BillingMeterNoMeterFoundEvent,
+    V1BillingMeterNoMeterFoundEventNotification as V1BillingMeterNoMeterFoundEventNotification,
 )
 from stripe.events._v2_core_event_destination_ping_event import (
     V2CoreEventDestinationPingEvent as V2CoreEventDestinationPingEvent,
+    V2CoreEventDestinationPingEventNotification as V2CoreEventDestinationPingEventNotification,
 )

stripe/events/_event_classes.py

@@ -1,18 +1,34 @@
 # -*- coding: utf-8 -*-
 # File generated from our OpenAPI spec
+from typing import Union
 from stripe.events._v1_billing_meter_error_report_triggered_event import (
     V1BillingMeterErrorReportTriggeredEvent,
+    V1BillingMeterErrorReportTriggeredEventNotification,
 )
 from stripe.events._v1_billing_meter_no_meter_found_event import (
     V1BillingMeterNoMeterFoundEvent,
+    V1BillingMeterNoMeterFoundEventNotification,
 )
 from stripe.events._v2_core_event_destination_ping_event import (
     V2CoreEventDestinationPingEvent,
+    V2CoreEventDestinationPingEventNotification,
 )
 
 
-THIN_EVENT_CLASSES = {
+V2_EVENT_CLASS_LOOKUP = {
     V1BillingMeterErrorReportTriggeredEvent.LOOKUP_TYPE: V1BillingMeterErrorReportTriggeredEvent,
     V1BillingMeterNoMeterFoundEvent.LOOKUP_TYPE: V1BillingMeterNoMeterFoundEvent,
     V2CoreEventDestinationPingEvent.LOOKUP_TYPE: V2CoreEventDestinationPingEvent,
 }
+
+V2_EVENT_NOTIFICATION_CLASS_LOOKUP = {
+    V1BillingMeterErrorReportTriggeredEventNotification.LOOKUP_TYPE: V1BillingMeterErrorReportTriggeredEventNotification,
+    V1BillingMeterNoMeterFoundEventNotification.LOOKUP_TYPE: V1BillingMeterNoMeterFoundEventNotification,
+    V2CoreEventDestinationPingEventNotification.LOOKUP_TYPE: V2CoreEventDestinationPingEventNotification,
+}
+
+ALL_EVENT_NOTIFICATIONS = Union[
+    V1BillingMeterErrorReportTriggeredEventNotification,
+    V1BillingMeterNoMeterFoundEventNotification,
+    V2CoreEventDestinationPingEventNotification,
+]