# Contains functions called by the top-level view functions # that process the user's request and return a rendered XML # template from typing import Optional from fastapi import Request, Query, Response from .enums import * from .diagnostics import Diagnostic from .config import ResourceConfig def fatal_response(operation: Operation, version: SRUVersion, diagnostics: list[Diagnostic], request, templates): """ Return a response with the fatal diagnostics and no other payload. """ diagStr = [str(d) for d in diagnostics] if operation in (Operation.explain, Operation.scan): if version == SRUVersion.v1_2: templateVersion = 1 else: templateVersion = 2 return templates.TemplateResponse('explain_response.xml', { 'request': request, 'diagnostics': diagStr, 'version': templateVersion }, media_type='application/xml') elif operation == Operation.searchRetrieve: return templates.TemplateResponse('search_retrieve_response.xml', { 'request': request, 'diagnostics': diagStr, 'n_hits': 0 }, media_type='application/xml') def process_explain(version: SRUVersion, searchOptions: dict[str, str], config: Optional[ResourceConfig], diagnostics: list[Diagnostic], request, templates): """ Process an explain request. Return a rendered XML response. """ if version == SRUVersion.v1_2: templateVersion = 1 else: templateVersion = 2 endpointDescNeeded = False if 'x-fcs-endpoint-description' in searchOptions and searchOptions['x-fcs-endpoint-description'] == 'true': endpointDescNeeded = True diagStr = [str(d) for d in diagnostics] return templates.TemplateResponse('explain_response.xml', { 'request': request, 'diagnostics': diagStr, 'config': config.as_dict(), 'version': templateVersion, 'endpoint_desc_needed': endpointDescNeeded }, media_type='application/xml') def process_search_retrieve(version: SRUVersion, queryType: QueryType, query: str, searchOptions: dict[str, str], config: Optional[ResourceConfig], diagnostics: list[Diagnostic], app, request, templates): """ Process a searchRetrieve request. Return a rendered XML response. """ if config.platform == CorpPlatform.tsakorpus: try: if queryType == QueryType.cql: strGetParams = app.qp_tsakorpus.translate_simple(query, config) else: strGetParams = app.qp_tsakorpus.translate_advanced(query, config) print(strGetParams) res = app.qp_tsakorpus.send_query(strGetParams, config) except Diagnostic as diag: return fatal_response(Operation.searchRetrieve, version, diagnostics + [diag], request, templates) records, nHits, diagnostics = app.rp_tsakorpus.parse(res, config, searchOptions['x-fcs-dataviews']) records = [r.as_dict() for r in records] diagnostics = [str(d) for d in diagnostics] return templates.TemplateResponse('search_retrieve_response.xml', { 'request': request, 'n_hits': nHits, 'records': records, 'diagnostics': diagnostics }, media_type='application/xml') elif config.platform == CorpPlatform.litterae: try: if queryType == QueryType.cql: strGetParams = app.qp_litterae.translate_simple(query, config) else: strGetParams = app.qp_litterae.translate_simple(query, config) # print(strGetParams) res = app.qp_litterae.send_query(strGetParams, config) print(res) except Diagnostic as diag: return fatal_response(Operation.searchRetrieve, version, diagnostics + [diag], request, templates) for dv in searchOptions['x-fcs-dataviews'].split(','): dv = dv.strip() if dv != 'hits' and version == SRUVersion.v2_0: # Litterae does not provide any additional annotation, so only Generic Hits # are available as a data view. # If SRU 1.2 is used, such a diagnostic has already been added # at a previous step. diagnostics.append(Diagnostic(DiagnosticTypes.fcs, 4, details=dv)) records, nHits, diagnostics = app.rp_litterae.parse(res, config, searchOptions['x-fcs-dataviews']) records = [r.as_dict() for r in records] diagnostics = [str(d) for d in diagnostics] return templates.TemplateResponse('search_retrieve_response.xml', { 'request': request, 'n_hits': nHits, 'records': records, 'diagnostics': diagnostics }, media_type='application/xml') def process_request(operation: Operation, version: SRUVersion, queryType: QueryType, query: str, searchOptions: dict[str, str], config: Optional[ResourceConfig], diagnostics: list[Diagnostic], app, request, templates): """ Process validated user request that came in through the endpoint() function in main.py. Return a rendered template. :param diagnostics: List of diagnostics produced by the validation function. """ print(query) # If something is clearly wrong with the query, return # a response with the list of diagnostics if config is None or any(d.is_fatal() for d in diagnostics): return fatal_response(operation, version, diagnostics, request, templates) # If everything looks good, proceed to query parsing if operation == Operation.searchRetrieve: return process_search_retrieve(version, queryType, query, searchOptions, config, diagnostics, app, request, templates) elif operation == Operation.explain: return process_explain(version, searchOptions, config, diagnostics, request, templates) # We should not end up here, but if we did, something went wrong and # no fatal diagnostic describes the problem. Add a generic fatal diagnostic # and return a fatal response. diagnostics.append(Diagnostic(DiagnosticTypes.sru, 1)) return fatal_response(operation, version, diagnostics, request, templates) if __name__ == '__main__': pass