# -*- 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 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} 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.parseEventNotification() 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) # 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, 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, 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, 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, usage=["fetch_related_object", "unknown_event"], ) return self._client.deserialize( response, api_mode=get_api_mode(self.related_object.url), )