from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import List
[docs]
class MarketplaceProduct:
def __init__(self, name: str, extension_list: List[str], product_id: str):
self.name = name
self.extension_list = extension_list
self.id = product_id
def __eq__(self, other):
return vars(self) == vars(other)
def __repr__(self):
return "<Object MarketplaceProduct: {}>".format(vars(self))
[docs]
@dataclass
class SubscriptionEvent:
"""Data structure for subscription related events
:param event: The payload received from the marketplace
:param event_type: Will be either SUBSCRIPTION_ORDER, SUBSCRIPTION_CANCEL, or SUBSCRIPTION_CHANGE
:param user_id: The UserID within the Q2 database
:param fi_name: The name of the Financial Institution
:param subscription_id: A guid that identifies the subscription
:param sso_user_id: A guid that represents this user on the marketplace
:param product_identifier: A unique identifer that represents a marketplace product
:param extension_url: The url that will be called for each subscription event
:param extension_name: The name of the installed extension that will be called for each subscription event
:param created: Iso Timestamp of product creation in the marketplace
:param processed: Iso Timestamp of product processing in the marketplace
.. code-block:: javascript
// Sample subscription event
{
"event": {
"type": "SUBSCRIPTION_CANCEL",
"marketplace": {
"partner": "Q2TOTALACCESS",
"baseUrl": "https://testq2totalaccess.appdirect.com"
},
"creator": {
"uuid": "4fc5fb4b-9809-49b5-abc5-7dab4544e685",
"openId": "https://testq2totalaccess.appdirect.com/openid/id/4fc5fb4b-9809-49b5-abc5-7dab4544e685",
"email": "jdoe@q2ebanking.com",
"firstName": "John",
"lastName": "Doe",
"language": "en",
"locale": "en-US",
"address": {
"salutation": null,
"firstName": "John",
"lastName": "Doe",
"fullName": "John Doe",
"companyName": null,
"phone": null,
"phoneExtension": null,
"fax": null,
"faxExtension": null,
"street1": "123 Some St",
"street2": null,
"city": "COLORADO SPRINGS",
"state": "CO",
"zip": "80909-5448",
"country": "US",
"pobox": null,
"pozip": null
},
"attributes": null
},
"payload": {
"user": null,
"company": null,
"account": {
"accountIdentifier": null,
"status": "ACTIVE",
"parentAccountIdentifier": null
},
"addonInstance": null,
"addonBinding": null,
"order": null,
"notice": null,
"configuration": {
"userExternalIdentifier": "169ac613-e9ac-433c-893c-15fffddeef49",
"applicationId": "ef02eb9b-0a4c-4be5-b78e-4088d7a8a55c",
"subscriptionId": "950cfd23-d730-4505-b538-d14a70a7a1b0"
}
},
"links": []
},
"event_type": "SUBSCRIPTION_CANCEL",
"user_id": 12345,
"fi_name": "My FI Name",
"subscription_id": "a3c9c831-e550-428f-b89c-43c5e7b338b9",
"sso_user_id": "b584a5f9-85a4-4653-90e6-8bbd41d71403",
"product_identifier": "",
"extension_url": "https://someurl/marketplace/my-product-name",
"created": "2021-12-06T15:27:32.636332",
"processed": "2021-12-06T15:27:32.636332"
}
"""
event: dict
event_type: str
user_id: int
subscription_id: str
sso_user_id: str
product_identifier: str
extension_url: str
extension_name: str = None
created: datetime = None
processed: datetime = None
fi_name: str = None
[docs]
@staticmethod
def from_json(body: dict):
created = body.get("created")
if created:
created = datetime.fromisoformat(created)
processed = body.get("processed")
if processed:
processed = datetime.fromisoformat(processed)
return SubscriptionEvent(
body["event"],
body["event_type"],
body["user_id"],
body["subscription_id"],
body["sso_user_id"],
body["product_identifier"],
body["extension_url"],
extension_name=body.get("extension_name"),
created=created,
processed=processed,
fi_name=body["fi_name"],
)
[docs]
class ErrorCode(Enum):
"""
Enumeration of error codes that can be returned during a SUBSCRIPTION_ORDER, SUBSCRIPTION_CANCEL, or SUBSCRIPTION_UPDATE event
"""
USER_ALREADY_EXISTS = "USER_ALREADY_EXISTS"
USER_NOT_FOUND = "USER_NOT_FOUND"
ACCOUNT_NOT_FOUND = "ACCOUNT_NOT_FOUND"
MAX_USERS_REACHED = "MAX_USERS_REACHED"
UNAUTHORIZED = "UNAUTHORIZED"
OPERATION_CANCELLED = "OPERATION_CANCELLED"
CONFIGURATION_ERROR = "CONFIGURATION_ERROR"
INVALID_RESPONSE = "INVALID_RESPONSE"
PENDING = "PENDING"
FORBIDDEN = "FORBIDDEN"
BINDING_NOT_FOUND = "BINDING_NOT_FOUND"
TRANSPORT_ERROR = "TRANSPORT_ERROR"
UNKNOWN_ERROR = "UNKNOWN_ERROR"
DO_NOT_CANCEL = "DO_NOT_CANCEL"
[docs]
class SubscriptionResult:
def __init__(self, success: bool):
self.success = success
[docs]
def to_json(self):
return vars(self)
def __repr__(self):
return "<Object SubscriptionResult: {}>".format(vars(self))
[docs]
class SubscriptionSuccess(SubscriptionResult):
def __init__(self, account_identifier: str = ""):
super().__init__(True)
self.account_identifier = account_identifier
[docs]
class SubscriptionFailure(SubscriptionResult):
def __init__(self, message: str, error_code: ErrorCode):
super().__init__(False)
if isinstance(error_code, str):
try:
error_code = ErrorCode(error_code)
except ValueError:
error_code = ErrorCode.UNKNOWN_ERROR
self.error_code = error_code.value
self.message = message