from typing import Optional
from q2_sdk.core import configuration
from q2_sdk.core.exceptions import ConfigurationError
from q2_sdk.core.http_handlers.hq_handler import Q2HqRequestHandler
from q2_sdk.hq.db.user import User
from q2_sdk.hq.models.account import Account
from q2_sdk.hq.models.account_list import AccountList
from q2_sdk.hq.models.online_user import OnlineUser
from q2_sdk.models.version import Version
from q2_sdk.tools.utils import to_bool
[docs]
class Q2OnlineRequestHandler(Q2HqRequestHandler):
"""
RequestHandler meant to be used for requests incoming from Online.
"""
HQ_AUTH_TOKEN_TYPE = "SlidingTimed"
AUTH_TOKEN_LIFE_IN_MINUTES = 15
def __init__(self, application, request, **kwargs):
super().__init__(application, request, **kwargs)
self._account_model_proxy = Account
self.online_user: Optional[OnlineUser] = None
self.account_list = AccountList[Account]()
@property
def default_summary_log(self):
summary_log = super().default_summary_log
summary_log["HQ_ID"] = self.online_user.hq_session_id
summary_log["login_name"] = self.online_user.login_name
return summary_log
async def _build_models_from_hq(self, element):
await super()._build_models_from_hq(element)
try:
self.online_user = OnlineUser(
element, customer_key=self.hq_credentials.customer_key
)
except AttributeError:
self.logger.debug("No online user could be found in request")
for account_elem in element.findall(".//AccountListResponseRecord"):
possible_data_elems = element.findall(".//Q2_AccountDataElements")
account = self._account_model_proxy(account_elem, possible_data_elems)
self.account_list.append(account)
@property
def smart(self):
try:
from q2_smart.api import SmartAPI, SmartConfig
from q2_smart.version import __version__ as SmartVersion
except ModuleNotFoundError as err: # pragma: no cover
raise ConfigurationError(
"q2-smart package must be installed to use Smart functionality"
) from err
settings = configuration.get_settings()
smart_url = settings.SMART_URL
smart_token = settings.SMART_TOKEN
if not smart_url:
raise ConfigurationError(
"Q2SDK_SMART_URL environment variable not detected"
)
elif not smart_token:
raise ConfigurationError("SMART_TOKEN not present in vault")
if Version(SmartVersion) > Version("0.1.0"):
smart_config = SmartConfig(self.online_user.user_id, self.hq_credentials)
else:
smart_config = SmartConfig(self.online_user.user_id)
return SmartAPI(self.logger, smart_url, smart_config, smart_token)
@property
def sentinel(self):
try:
from q2_sentinel.api import SentinelAPI, SentinelConfig
except ModuleNotFoundError as err: # pragma: no cover
raise ConfigurationError(
"q2-sentinel package must be installed to use Sentinel functionality"
) from err
sentinel_config = SentinelConfig(
self.online_user.user_id,
self.online_user.login_name,
self.online_session.session_id,
)
return SentinelAPI(self.logger, self.hq_credentials, config=sentinel_config)
async def get_user_role(self):
if not self.online_user.user_role_id:
user = (
await User(self.logger, hq_credentials=self.hq_credentials).get(
user_id=self.online_user.user_id
)
)[0]
role_id = user.findtext("UserRoleID", "")
self.online_user.user_role_id = role_id
if role_id:
self.online_user.as_demographic_info().user_role_id = role_id
else:
self.logger.debug("UserRoleID is already populated, skipping DB call")
role_id = self.online_user.user_role_id
return role_id
async def validate_is_company(self) -> bool:
if not self.online_user.is_company:
customer = (
await self.db.customer.get(customer_id=self.online_user.customer_id)
)[0]
is_company = to_bool(customer.findtext("IsCompany", "false"))
self.online_user.is_company = is_company
else:
self.logger.debug("IsCompany is already populated, skipping DB call")
is_company = self.online_user.is_company
return is_company
async def validate_is_treasury(self) -> bool:
if not self.online_user.is_treasury:
group = await self.db.group.get_by_id(self.online_user.group_id)
is_treasury = to_bool(group.findtext("IsTreasury", "false"))
self.online_user.is_treasury = is_treasury
else:
self.logger.debug("IsTreasury is already populated, skipping DB call")
is_treasury = self.online_user.is_treasury
return is_treasury
async def validate_is_commercial(self) -> bool:
if not self.online_user.is_commercial:
group = await self.db.group.get_by_id(self.online_user.group_id)
is_commercial = to_bool(group.findtext("IsCommercial", "false"))
self.online_user.is_commercial = is_commercial
else:
self.logger.debug("IsCommercial is already populated, skipping DB call")
is_commercial = self.online_user.is_commercial
return is_commercial