I'm currently making an application to get definition of datapoints. There are three files involved. UniverseService.py get all the datapoints related to a datapoint's name. get_definition.py will get a definition of a datapoint. DataPointService will be the endpoint where it gets all the definition of those datapoints all at once. Because some datapoint might have 40-50 related data points which equal that amount of api calls. I would like to do async on the calls but I'm not sure how. I've been trying several ways with asyncio but it still doesn't work. The end goal is to import DataPointService as ds in a jupyter notebook and just call a function.
This is DataPointService
This is DataPointService
async def get_related_datapoints_definition(universe, name, quantity=None):
universeService = UniverseService(universe)
# Asynchronously find similar data points using run_in_executor
loop = asyncio.get_event_loop()
data_point_list = await loop.run_in_executor(None, universeService.find_similar_datapoint, name, quantity)
# Asynchronously get universe datapoints name-ID pairs using run_in_executor
pairs = await loop.run_in_executor(None, universeService.get_universe_datapoints_nameID_pairs)
# Use asyncio.gather to fetch the definitions concurrently
tasks = [get_datapoint_definition_async(universeService, pairs[data_point][0]) for data_point in data_point_list]
result = await asyncio.gather(*tasks)
return result
async def get_datapoint_definition_async(universeService, datapoint_id):
# Asynchronously get the definition for the given data point ID
definition = await universeService.get_datapoint_definition_by_id(datapoint_id)
return definitionThis is get_definition.pydef get_definition_url(datapoint_id,):
datapoint_definition=requests.get(
"abc",
headers={
'Authorization': f'Bearer {abc}',
'X-API-ProductId': "abc"
},
verify=False
)
url = datapoint_definition.json()
if type(url) is list: # if there is a definition, the data will be return in list
return url[0]["abc"]
else: # return in dictionary like {"message":"no data point defition"}
return -1
def get_definition(url):
if url == -1:
return "There is no definition for this Data Point"
definition = requests.get(url)
soup = BeautifulSoup(definition.content,'html.parser') #Parse html data
try:
# Attempt to find <p> element
p_element = soup.find('p')
if p_element:
# Case 1: Only <p> tag is found
# Check if the <p> element contains a <span> element
target_element = p_element.find('span')
if target_element:
target_element = p_element.find('span')
else:
target_element = p_element
else:
# Case 3: Only <body> tag is found
target_element = soup.find('body')
pass # Do something with target_element
except AttributeError:
# Handle any other exceptions or errors here
target_element = None # or any other default value you want to assign
"""s
Split text since the first word and the second always separate by a new line
"""
definition_text = target_element.get_text(strip=True)
print(definition_text)
# definition_split = definition_text.split(" ",1)
# first_word = " ".join(definition_split[0].split("\r\n"))
# parsed_definition = first_word + " " + definition_split[1]
return definition_textthis is UniverseService async def get_datapoint_definition_by_id(self, id):
url = await get_definition_url(id)
definition = await get_definition(url)
return definition