import json
from q2_sdk.models.tecton import InternalServerError
from q2_sdk.core.http_handlers.online_handler import Q2OnlineRequestHandler
from q2_sdk.core.http_handlers.sso_handler import (
Q2SSORequestHandler,
Q2SSOResponse,
ResponseType,
)
from q2_sdk.core.http_handlers.tecton_client_handler import Q2TectonClientRequestHandler
from q2_sdk.core.http_handlers.tecton_server_handler import Q2TectonServerRequestHandler
from q2_sdk.ui import forms, modals
from q2_marketplace import configuration, subscription
[docs]
class Q2MarketplaceMixin:
"""
Meant to be used in conjunction with a Q2BaseRequestHandler:
i.e. class Handler(Q2MarketplaceMixin, Q2BaseHandler):
...
Adds validation on checks on Marketplace subscription status
"""
MARKETPLACE_PRODUCTS = configuration.marketplace_products
SUBSCRIPTION_ERROR_TITLE = "Subscription"
SUBSCRIPTION_ERROR_MESSAGE = "Your subscription is not valid for this product. Please contact your administrator."
async def q2_post(self, *args, **kwargs):
overrides = self.db_config.get("_overrides", {})
skip_conditions = (
self.form_fields.get("routing_key") in self._unparsed_routes,
overrides.get("ignore_subscription") is True,
)
if any(skip_conditions):
self.logger.debug("Bypassing marketplace check because of DB override")
return await super().q2_post(*args, **kwargs)
validate = "validate"
failure = "failure"
handler_mapping = {
# Order is important here as the Tecton handlers inherit from online!
Q2TectonClientRequestHandler: {
validate: self.validate_subscription,
failure: self._tecton_client_failure,
},
Q2TectonServerRequestHandler: {
validate: self.validate_subscription,
failure: self._tecton_server_failure,
},
Q2OnlineRequestHandler: {
validate: self.validate_subscription,
failure: self._online_failure,
},
Q2SSORequestHandler: {
validate: self.validate_subscription,
failure: self._sso_failure,
},
}
if self.MARKETPLACE_PRODUCTS:
for handler_type, funcs in handler_mapping.items():
if isinstance(self, handler_type):
is_valid = await funcs[validate]()
if not is_valid:
return await funcs[failure]()
return await super().q2_post(*args, **kwargs)
[docs]
async def validate_subscription(self) -> bool:
"""Checks this user's subscription status for this product within the installed marketplace.
Returns true if the subscription is considered valid.
"""
is_valid = await subscription.is_valid(
self.logger,
self.extension_name,
self.online_user.user_id,
self.hq_credentials,
)
return is_valid
async def _online_failure(self):
return forms.Q2Form(
header="",
modal=modals.ErrorModal(
self.SUBSCRIPTION_ERROR_TITLE,
self.SUBSCRIPTION_ERROR_MESSAGE,
show_close=False,
modal_actions=modals.ModalAction(
"Close", external_route="/landingPage"
),
),
hide_submit_button=True,
)
async def _sso_failure(self):
return Q2SSOResponse(
ResponseType.HTML, response=self.SUBSCRIPTION_ERROR_MESSAGE
)
async def _tecton_client_failure(self):
return json.dumps(
InternalServerError(
self.SUBSCRIPTION_ERROR_TITLE,
{"message": self.SUBSCRIPTION_ERROR_MESSAGE},
).to_json()
)
async def _tecton_server_failure(self):
return forms.Q2TectonForm(
"",
[{"key": "q2token", "value": self.online_session.session_id}],
extension_name=self.extension_name,
tecton_version=self.TECTON_URL,
head_content=self.SUBSCRIPTION_ERROR_MESSAGE,
foot_content="",
hide_submit_button=True,
base_assets_url=self.base_assets_url,
)