Commit 94844fc1 authored by David Mendez's avatar David Mendez
Browse files

Add context loader module;

parent 84154930
Module that loads contexts from results
import requests
from app.config import RUN_CONFIG
from app.cache import CACHE
class ContextLoaderError(Exception):
"""Base class for exceptions in this module."""
WEB_RESULTS_SIZE_LIMIT = RUN_CONFIG.get('filter_query_max_clauses')
def get_context_url(context_dict):
returns the url for loading the context
:param context_dict: dict describing the context
return f'{context_dict["delayed_jobs_base_url"]}/outputs/{context_dict["context_id"]}/results.json'
def get_context(context_dict):
Returns the context described by the context dict
:param context_dict: dictionary describing the context
:return: the context loaded as an object
context_url = get_context_url(context_dict)
context_request = requests.get(context_url)
if context_request.status_code != 200:
raise ContextLoaderError('There was an error while loading the context: ' + context_request.text)
results = context_request.json()['search_results']
total_results = len(results)
if total_results > WEB_RESULTS_SIZE_LIMIT:
results = results[0:WEB_RESULTS_SIZE_LIMIT]
return results, total_results
def load_context_index(context_id, id_property, context):
Loads an index based on the id property of the context, for fast access
:param context_id: id of the context loaded
:param id_property: property used to identify each item
:param context: context loaded
context_index_key = 'context_index-{}'.format(context_id)
context_index = CACHE.get(key=context_index_key)
if context_index is None:
context_index = {}
for index, item in enumerate(context):
context_index[item[id_property]] = item
context_index[item[id_property]]['index'] = index
CACHE.set(key=context_index_key, value=context_index, timeout=3600)
return context_index
Module to test the context loader
import unittest
from app.blueprints.es_proxy.controllers.helpers import context_loader
class TestContextLoader(unittest.TestCase):
Class to test the functions used in the context loader
def test_generates_the_correct_context_url(self):
Tests that generates the initial url to load a context
context_dict = {
'delayed_jobs_base_url': '',
'context_type': 'SIMILARITY',
'context_id': 'STRUCTURE_SEARCH-V4s_piFIe9FL7sOuJ0jl6U0APMEsoV-b4HfFE_dtojc='
context_url_must_be = f'{context_dict["delayed_jobs_base_url"]}' \
context_url_got = context_loader.get_context_url(context_dict)
self.assertEqual(context_url_must_be, context_url_got, msg='The context url was not generated correctly!')
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment