# 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 initial_validation(operation, version, queryType, searchOptions, query):
"""
Validate and convert values of the main request parameters.
Return converted values and a list of fatal diagnostics, if anything is wrong.
"""
failDiagnoctics = []
if version == '1.2':
version = SRUVersion.v1_2
elif version == '2.0':
version = SRUVersion.v2_0
else:
version = SRUVersion.v2_0
failDiagnoctics.append(Diagnostic(DiagnosticTypes.sru, 5, details='2.0'))
if operation == '':
if len(query) > 0:
operation = Operation.searchRetrieve
else:
operation = Operation.explain
elif operation == 'explain':
operation = Operation.explain
elif operation == 'searchRetrieve':
operation = Operation.searchRetrieve
elif operation == 'scan':
operation = Operation.scan
else:
operation = Operation.explain
failDiagnoctics.append(Diagnostic(DiagnosticTypes.sru, 4, version=version))
if queryType == 'fcs':
queryType = QueryType.fcs
elif queryType == 'cql':
queryType = QueryType.cql
else:
queryType = QueryType.cql
failDiagnoctics.append(Diagnostic(DiagnosticTypes.sru, 6, message='Supported query types: fcs and cql.',
version=version))
try:
searchOptions['startRecord'] = int(searchOptions['startRecord'])
except ValueError:
searchOptions['startRecord'] = 1
failDiagnoctics.append(Diagnostic(DiagnosticTypes.sru, 6, message='startRecord should be a positive integer.',
version=version))
if searchOptions['startRecord'] < 1:
failDiagnoctics.append(Diagnostic(DiagnosticTypes.sru, 6, message='startRecord should be a positive integer.',
version=version))
try:
searchOptions['maximumRecords'] = int(searchOptions['maximumRecords'])
except ValueError:
searchOptions['maximumRecords'] = 0
failDiagnoctics.append(
Diagnostic(DiagnosticTypes.sru, 6, message='maximumRecords should be a non-negative integer.',
version=version))
if searchOptions['maximumRecords'] < 0:
failDiagnoctics.append(
Diagnostic(DiagnosticTypes.sru, 6, message='maximumRecords should be a non-negative integer.',
version=version))
# recordPacking has entirely different semantics in SRU 1.2 and SRU 2.0
if version == SRUVersion.v1_2:
if searchOptions['recordPacking'] == '':
searchOptions['recordPacking'] = 'xml'
if searchOptions['recordPacking'] not in ('xml', 'string'):
failDiagnoctics.append(
Diagnostic(DiagnosticTypes.sru, 71, message='recordPacking should equal "xml" or "string".',
version=version))
else:
if searchOptions['recordXMLEscaping'] == '':
searchOptions['recordXMLEscaping'] = 'xml'
if searchOptions['recordPacking'] == '':
searchOptions['recordPacking'] = 'packed'
if searchOptions['recordXMLEscaping'] not in ('xml', 'string'):
failDiagnoctics.append(
Diagnostic(DiagnosticTypes.sru, 71, message='recordXMLEscaping should equal "xml" or "string".',
version=version))
if searchOptions['recordPacking'] not in ('packed', 'unpacked'):
failDiagnoctics.append(
Diagnostic(DiagnosticTypes.sru, 6, message='recordPacking should equal "packed" or "unpacked".',
version=version))
try:
searchOptions['resultSetTTL'] = int(searchOptions['resultSetTTL'])
except ValueError:
searchOptions['resultSetTTL'] = 0
failDiagnoctics.append(
Diagnostic(DiagnosticTypes.sru, 6, message='resultSetTTL should be a positive integer.',
version=version))
if searchOptions['resultSetTTL'] < 0:
# This does not look good, but we don't care because this
# value is not used anyway
pass
# failDiagnoctics.append(
# Diagnostic(DiagnosticTypes.sru, 6, message='resultSetTTL should be a positive integer.',
# version=version))
return operation, version, queryType, searchOptions, failDiagnoctics
def fatal_response(operation: Operation,
version: SRUVersion,
config: Optional[ResourceConfig],
diagnostics: list[Diagnostic],
request, templates):
"""
Return a response with the fatal diagnostics
and no other payload.
"""
if config is None:
configStr = ''
else:
configStr = config.as_dict()
if version == SRUVersion.v1_2:
templateVersion = 1
else:
templateVersion = 2
for diag in diagnostics:
diag.version = version
diagStr = [str(d) for d in diagnostics]
if operation in (Operation.explain, Operation.scan):
return templates.TemplateResponse('explain_response.xml',
{
'request': request,
'diagnostics': diagStr,
'config': configStr,
'version': templateVersion
},
media_type='application/xml')
elif operation == Operation.searchRetrieve:
return templates.TemplateResponse('search_retrieve_response.xml',
{
'request': request,
'diagnostics': diagStr,
'n_hits': 0,
'version': templateVersion
},
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
for diag in diagnostics:
diag.version = version
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,
config: Optional[ResourceConfig],
diagnostics: list[Diagnostic],
app, request, templates):
"""
Process a searchRetrieve request.
Return a rendered XML response.
"""
if version == SRUVersion.v1_2:
templateVersion = 1
else:
templateVersion = 2
for diag in diagnostics:
diag.version = version
if config.platform == CorpPlatform.annis:
try:
if queryType == QueryType.cql:
query = app.qp_annis.translate_simple(query, config, searchOptions)
else:
query = app.qp_annis.translate_advanced(query, config, searchOptions)
print(query)
res = app.qp_annis.send_query(query, config)
except Diagnostic as diag:
return fatal_response(Operation.searchRetrieve, version, config, diagnostics + [diag], request, templates)
# return query['query']
return res
# records, nHits, diagnostics = app.rp_annis.parse(res, config, searchOptions['x-fcs-dataviews'])
# if any(diag.is_fatal() for diag in diagnostics):
# return fatal_response(Operation.searchRetrieve, version, config, diagnostics, request, templates)
# 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,
# 'version': templateVersion,
# 'diagnostics': diagnostics
# },
# media_type='application/xml')
if config.platform == CorpPlatform.tsakorpus:
try:
if queryType == QueryType.cql:
strGetParams = app.qp_tsakorpus.translate_simple(query, config, searchOptions)
else:
strGetParams = app.qp_tsakorpus.translate_advanced(query, config, searchOptions)
print(strGetParams)
res = app.qp_tsakorpus.send_query(strGetParams, config)
except Diagnostic as diag:
return fatal_response(Operation.searchRetrieve, version, config, diagnostics + [diag], request, templates)
records, nHits, diagnostics = app.rp_tsakorpus.parse(res, config, searchOptions)
if any(diag.is_fatal() for diag in diagnostics):
return fatal_response(Operation.searchRetrieve, version, config, diagnostics, request, templates)
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,
'version': templateVersion,
'diagnostics': diagnostics
},
media_type='application/xml')
elif config.platform == CorpPlatform.litterae:
try:
if queryType == QueryType.cql:
strGetParams = app.qp_litterae.translate_simple(query, config, searchOptions)
else:
# No advanced search for Litterae
strGetParams = app.qp_litterae.translate_simple(query, config, searchOptions)
# print(strGetParams)
res = app.qp_litterae.send_query(strGetParams, config)
print(res)
except Diagnostic as diag:
return fatal_response(Operation.searchRetrieve, version, config, 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, version=version))
records, nHits, diagnostics = app.rp_litterae.parse(res, config, searchOptions)
if any (diag.is_fatal() for diag in diagnostics):
return fatal_response(Operation.searchRetrieve, version, config, diagnostics, request, templates)
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,
'version': templateVersion,
'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, config, 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, version=version))
return fatal_response(operation, version, config, diagnostics, request, templates)
if __name__ == '__main__':
pass