import functools
from typing import List, Optional
from q2_sdk.hq.models.hq_credentials import HqCredentials
from ..models.core_user import CoreUser
from ..queries.base_query import BaseQuery
[docs]
class BaseMapper:
"""
All mappers will inherit from this common base class.
Typically not used directly, but in conjunction with a
BaseCore that invokes these methods.
"""
def __init__(
self,
list_of_queries: List[BaseQuery],
hq_credentials: Optional[HqCredentials] = None,
zone_context: Optional[CoreUser] = None,
):
assert isinstance(list_of_queries, list)
self.hq_credentials = hq_credentials
self.list_of_queries = list_of_queries
self.zone_context = zone_context
async def _run_queries(self):
for query in self.list_of_queries:
func = functools.partial(query.execute, self.hq_credentials)
if self.zone_context:
func.keywords["zone_context"] = self.zone_context
await func()
def _parse_responses(self):
try:
return self.parse_returned_queries(self.list_of_queries)
except Exception:
log_data = [x.raw_core_response for x in self.list_of_queries]
self.list_of_queries[0].logger.error(
f"Problem mapping core response(s): {log_data}"
)
raise
[docs]
async def execute(self):
"""
- Calls .execute on all queries in self.list_of_queries
- Calls .parse_returned_queries on the results
"""
await self._run_queries()
return self._parse_responses()
[docs]
@staticmethod
def parse_returned_queries(list_of_queries):
"""
After all queries in list_of_queries have been run, this
parses the results into a standardized response type.
"""
raise NotImplementedError