1""" GenAPI """ 2import re 3 4import requests 5import requests_cache 6from genesis import GenData, Genesis 7 8from orangecontrib.bioinformatics.resolwe.utils import GENAPI_CACHE, CACHE_BACKEND, response_to_json 9 10view_model = ['experiment', 'growth', 'genotype', 'treatment', 'strain', 'time', 'replicate'] 11DEFAULT_EMAIL = 'anonymous@genialis.com' 12DEFAULT_PASSWD = 'anonymous' 13DEFAULT_URL = 'https://dictyexpress.research.bcm.edu' 14 15 16class GenAPI: 17 """Python module that leverages Genesis PyAPI (Python API for accsess to DictyExpress database). 18 19 It supports connection to the server and data retrieval functionalities. 20 """ 21 22 def __init__(self, email=DEFAULT_EMAIL, password=DEFAULT_PASSWD, url=DEFAULT_URL): 23 self._gen = Genesis(email, password, url) 24 self.email = email 25 self._data_endpoint = url + '/data/' 26 27 @staticmethod 28 def clear_cache(): 29 with requests_cache.enabled(cache_name=GENAPI_CACHE, backend=CACHE_BACKEND): 30 requests_cache.clear() 31 32 def get_cached_ids(self): 33 with requests_cache.enabled(cache_name=GENAPI_CACHE, backend=CACHE_BACKEND): 34 cached_object = requests_cache.core.get_cache() 35 responses = [cached_object.get_response_and_time(response) for response in cached_object.responses] 36 gen_ids = [] 37 38 for url in [response.url for response, _ in responses]: 39 gen_id = re.search(r'{}(.*?)/'.format(self._data_endpoint), url) 40 41 if gen_id is not None: 42 gen_ids.append(gen_id.group(1)) 43 44 return gen_ids 45 46 def fetch_etc_objects(self, *args, **kwargs): 47 """Function downloads all available :obj:`GenData` etc objects from DictyExpress database. 48 49 :rtype: list of GenData objects 50 """ 51 52 with requests_cache.enabled(cache_name=GENAPI_CACHE, backend=CACHE_BACKEND): 53 try: 54 # Note: this is hardcoded for now. When we port this module to Resolwe platform 55 # data retrieval will be different 56 list_of_experiments = self._gen.api.data.get( 57 case_ids__contains='5535115cfad58d5e03006217', status='done', type__startswith='data:etc:' 58 )['objects'] 59 60 except requests.exceptions.ConnectionError as e: 61 raise requests.exceptions.ConnectionError('Server not accessible, check your connection.') from e 62 63 store_experiments = [GenData(exp, self._gen) for exp in list_of_experiments] 64 65 return store_experiments 66 67 def download_etc_data(self, gen_data_id, *args, **kwargs): 68 """Function downloads etc data of a chosen experiment from the server. 69 70 :param gen_data_id: id of GeneData object 71 :type gen_data_id: str 72 73 :rtype: data in json like format 74 """ 75 table_name = kwargs.get("table_name", '') 76 77 with requests_cache.enabled(cache_name=GENAPI_CACHE, backend=CACHE_BACKEND): 78 try: 79 response = next(self._gen.download([gen_data_id], 'output.etcfile')) 80 81 except requests.exceptions.ConnectionError as e: 82 raise requests.exceptions.ConnectionError('Server not accessible, check your connection.') from e 83 84 return response_to_json(response), table_name 85