# -*- coding: utf-8 -*- import json from typing import Any, ClassVar, Dict, Optional, cast from typing_extensions import Literal, TYPE_CHECKING from stripe._stripe_object import StripeObject, UntypedStripeObject from stripe._util import get_api_mode from stripe._stripe_context import StripeContext if TYPE_CHECKING: from stripe._stripe_client import StripeClient # The beginning of the section generated from our OpenAPI spec class Event(StripeObject): """ Events are generated to keep you informed of activity in your business account. APIs in the /v2 namespace generate [thin events](https://docs.stripe.com/event-destinations#benefits-of-thin-events) which have small, unversioned payloads that include a reference to the ID of the object that has changed. The Events v2 API returns these new thin events. [Retrieve the event object](https://docs.stripe.com/event-destinations#fetch-data) for additional data about the event. Use the related object ID in the event payload to [fetch the API resource](https://docs.stripe.com/event-destinations#retrieve-the-object-associated-with-thin-events) of the object associated with the event. Comparatively, events generated by most API v1 include a versioned snapshot of an API object in their payload. """ OBJECT_NAME: ClassVar[Literal["v2.core.event"]] = "v2.core.event" class Reason(StripeObject): class Request(StripeObject): id: str """ ID of the API request that caused the event. """ idempotency_key: str """ The idempotency key transmitted during the request. """ request: Optional[Request] """ Information on the API request that instigated the event. """ type: Literal["request"] """ Event reason type. """ _inner_class_types = {"request": Request} changes: Optional[UntypedStripeObject[Any]] """ Before and after changes for the primary related object. """ context: Optional[str] """ Authentication context needed to fetch the event or related object. """ created: str """ Time at which the object was created. """ id: str """ Unique identifier for the event. """ livemode: bool """ Has the value `true` if the object exists in live mode or the value `false` if the object exists in test mode. """ object: Literal["v2.core.event"] """ String representing the object's type. Objects of the same type share the same value of the object field. """ reason: Optional[Reason] """ Reason for the event. """ type: str """ The type of the event. """ _inner_class_types = {"reason": Reason} # The end of the section generated from our OpenAPI spec class ReasonRequest: id: str idempotency_key: str def __init__(self, d) -> None: self.id = d["id"] self.idempotency_key = d["idempotency_key"] def __repr__(self) -> str: return f"" class Reason: type: Literal["request"] request: Optional[ReasonRequest] = None def __init__(self, d) -> None: self.type = d["type"] if self.type == "request": self.request = ReasonRequest(d["request"]) def __repr__(self) -> str: return f"" class RelatedObject: id: str type: str url: str def __init__(self, d) -> None: self.id = d["id"] self.type = d["type"] self.url = d["url"] def __repr__(self) -> str: return f"" class EventNotification: """ EventNotification represents the json that's delivered from an Event Destination. It's a basic struct-like object with a few convenience methods. Use `fetch_event()` to get the full event object. """ id: str """ Unique identifier for the event. """ type: str """ The type of the event. """ created: str """ Time at which the object was created. """ livemode: bool """ Livemode indicates if the event is from a production(true) or test(false) account. """ context: Optional[StripeContext] = None """ [Optional] Authentication context needed to fetch the event or related object. """ reason: Optional[Reason] = None """ [Optional] Reason for the event. """ def __init__( self, parsed_body: Dict[str, Any], client: "StripeClient" ) -> None: self.id = parsed_body["id"] self.type = parsed_body["type"] self.created = parsed_body["created"] self.livemode = bool(parsed_body.get("livemode")) context_value = parsed_body.get("context") if context_value: self.context = StripeContext.parse(context_value) if parsed_body.get("reason"): self.reason = Reason(parsed_body["reason"]) self._client = client @staticmethod def from_json(payload: str, client: "StripeClient") -> "EventNotification": """ Helper for constructing an Event Notification. Doesn't perform signature validation, so you should use StripeClient.parse_event_notification() instead for initial handling. This is useful in unit tests and working with EventNotifications that you've already validated the authenticity of. """ parsed_body = json.loads(payload) if parsed_body.get("object") == "event": raise ValueError( "You passed a webhook payload to StripeClient.parse_event_notification, which expects a thin event notification. Use StripeClient.construct_event instead." ) # circular import busting from stripe.events._event_classes import ( get_v2_event_notification_class, ) event_class = get_v2_event_notification_class(parsed_body["type"]) return event_class(parsed_body, client) def __repr__(self) -> str: return f"" def fetch_event(self) -> Event: response = self._client.raw_request( "get", f"/v2/core/events/{self.id}", stripe_context=self.context, headers={"Stripe-Request-Trigger": f"event={self.id}"}, usage=["pushed_event_pull"], ) return cast(Event, self._client.deserialize(response, api_mode="V2")) async def fetch_event_async(self) -> Event: response = await self._client.raw_request_async( "get", f"/v2/core/events/{self.id}", stripe_context=self.context, headers={"Stripe-Request-Trigger": f"event={self.id}"}, usage=["pushed_event_pull", "pushed_event_pull_async"], ) return cast(Event, self._client.deserialize(response, api_mode="V2")) class UnknownEventNotification(EventNotification): """ Represents an EventNotification payload that the SDK doesn't have types for. May have a related object. """ related_object: Optional[RelatedObject] = None """ [Optional] Object containing the reference to API resource relevant to the event. """ def __init__( self, parsed_body: Dict[str, Any], client: "StripeClient" ) -> None: super().__init__(parsed_body, client) if parsed_body.get("related_object"): self.related_object = RelatedObject(parsed_body["related_object"]) def fetch_related_object(self) -> Optional[StripeObject]: if self.related_object is None: return None response = self._client.raw_request( "get", self.related_object.url, stripe_context=self.context, headers={"Stripe-Request-Trigger": f"event={self.id}"}, usage=["fetch_related_object", "unknown_event"], ) return self._client.deserialize( response, api_mode=get_api_mode(self.related_object.url), ) async def fetch_related_object_async(self) -> Optional[StripeObject]: if self.related_object is None: return None response = await self._client.raw_request_async( "get", self.related_object.url, stripe_context=self.context, headers={"Stripe-Request-Trigger": f"event={self.id}"}, usage=["fetch_related_object", "unknown_event"], ) return self._client.deserialize( response, api_mode=get_api_mode(self.related_object.url), )