Source code for q2_cores.CoreAPI.queries.base_query

from enum import Enum
import logging
from pathlib import Path
from typing import Optional

from lxml import etree
from lxml.builder import E

from q2_sdk.models.cores.queries.base_query import BaseQuery

from q2_sdk.models.cores.queries.jinja_query import JinjaQuery


[docs] class LookupType(Enum): SOCIAL_SECURITY_NUMBER = "taxid" MEMBER_NUMBER = "MemberNumber"
[docs] class CoreAPIBaseQuery(BaseQuery): """ All CoreAPI queries inherit from this """ def __init__( self, logger: logging.Logger, request_type: str, req_data: dict, cif: Optional[str] = None, mock_failure=False, ): self.req_data = req_data self.request_type = request_type self.cif = cif super().__init__(logger, mock_failure)
[docs] def make_query(self, name: str, context: dict = None) -> JinjaQuery: path = Path(__file__).parent / "templates" request_template = f"{name}Request.xml" mock_response_template = f"{name}Response.xml" return JinjaQuery( logger=self.logger, templates_path=path, request_template=request_template, mock_response_template=mock_response_template, context=context, collapse_whitespace=True, )
[docs] def build(self) -> str: """ Constructs and returns the core request from a template and provided parameters. """ request = self.make_query(self.request_type, self.req_data) cif = self.cif if self.cif else "" request_xml = etree.tostring(E.root(taxid=cif, Request=request.build())) return request_xml.decode()
[docs] def mock_response(self): return self.make_query(self.request_type, self.req_data).mock_response()