import logging
from typing import List, Optional
from q2_sdk.hq.models.hq_credentials import HqCredentials
from q2_sdk.models.cores.mappers.base_mapper import BaseMapper
from q2_sdk.models.cores.queries.base_query import BaseQuery
from q2_cores.Corelation.models import LoginVerifyResponse
from q2_cores.Corelation.queries import LoginVerifyQuery
from q2_cores.exceptions import CoreException
[docs]
class LoginVerifyMapper(BaseMapper):
def __init__(
self,
list_of_queries: List[BaseQuery],
logger: logging.Logger,
hq_credentials: Optional[HqCredentials] = None,
):
super().__init__(list_of_queries, hq_credentials=hq_credentials)
self.logger = logger
[docs]
def parse_returned_queries(
self, list_of_queries: List[BaseQuery]
) -> LoginVerifyResponse:
assert len(list_of_queries) == 1
assert isinstance(list_of_queries[0], LoginVerifyQuery)
response = list_of_queries[0].raw_core_response
response_obj = LoginVerifyResponse(response)
if response_obj.error is not None:
raise CoreException(
f"There was a problem with the request to the core. Response: {response_obj.error}"
)
return response_obj