[Python] How to optimize my performances, using asynchronous python code

I'm looking to optimize my code in order to process the info faster. First time playing with asynchronous requests. And also still new to Python. I hope my code makes sense.

I'm using FastAPI as a framework. And aiohttp to send my requests. Right now, I'm only interested in getting the total of results per word searched. I will be dumping the json into a DB afterwards.

My code is sending requests to the public crossref API (crossref) As an example, I'm searching for the terms from 2022-06-02 to 2022-06-03 (inclusive). The terms being searched are: 'paper' (3146 results), 'ammonium' (1430 results) and 'bleach' (23 results). Example:

https://api.crossref.org/works?rows=1000&sort=created&mailto=youremail@domain.com&query=paper&filter=from-index-date:2022-06-02,until-index-date:2022-06-03&cursor=*


This returns 3146 rows. I need to search for only one term at a time. I did not try to split it per day as well to see if it's faster.

There is also a recursive context in this. This is where I feel like I'm mishandling the asynchronous concept. Here is why I need a recursive call.

Deep paging requests

Deep paging using cursors can be used to iterate over large result sets, without any limits on their size. To use deep paging make a query as normal, but include the cursor parameter with a value of *, for example:

https://api.crossref.org/works?rows=1000&sort=created&mailto=youremail@domain.com&query=ammonium&filter=from-index-date:2022-06-02,until-index-date:2022-06-03&cursor=*


A next-cursor field will be provided in the JSON response. To get the next page of results, pass the value of next-cursor as the cursor parameter. For example:

https://api.crossref.org/works?rows=1000&sort=created&mailto=youremail@domain.com&query=ammonium&filter=from-index-date:2022-06-02,until-index-date:2022-06-03&cursor=<value of next-cursor parameter>


Advice from the CrossRef doc

Clients should check the number of returned items. If the number of returned items is equal to the number of expected rows then the end of the result set has been reached. Using next-cursor beyond this point will result in responses with an empty items list.
​

My processing time is still through the roof with just 3 words (and 7 requests), it's over 15sec. I'm trying to turn that down to under 5 seconds if possible? Using postman, the longest request took about 4 seconds to come back

This is what I have so far if you want to try it out.

schema.py

class CrossRefSearchRequest(BaseModel):
keywords: List[str]
date_from: Optional[datetime] = None
date_to: Optional[datetime] = None


controler.py

import time
from fastapi import FastAPI, APIRouter, Request

app = FastAPI(title="CrossRef API", openapi_url=f"{settings.API_V1_STR}/openapi.json")
api_router = APIRouter()
service = CrossRefService()

@api_router.post("/search", status_code=201)
async def search_keywords(*, search_args: CrossRefSearchRequest) -> dict:
fixed_search_args = {
"sort": "created",
"rows": "1000",
"cursor": "*"
}
results = await service.cross_ref_request(search_args, **fixed_search_args)
return {k: len(v) for k, v in results.items()}

# sets the header X-Process-Time, in order to have the time for each request
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response

app.include_router(api_router)

if __name__ == "__main__":
# Use this for debugging purposes only
import uvicorn

uvicorn.run(app, host="0.0.0.0", port=8001, log_level="debug")


service.py

from datetime import datetime, timedelta

def _setup_date_default(date_from_req: datetime, date_to_req: datetime):
yesterday = datetime.utcnow()- timedelta(days=1)

date_from = yesterday if date_from_req is None else date_from_req
date_to = yesterday if date_to_req is None else date_to_req

return date_from.strftime(DATE_FORMAT_CROSS_REF), date_to.strftime(DATE_FORMAT_CROSS_REF)


class CrossRefService:

def __init__(self):
self.client = CrossRefClient()

# my recursive call for the next cursor
async def _send_client_request(self ,final_result: dict[str, list[str]], keywords: [str], date_from: str, date_to: str, **kwargs):
json_responses = await self.client.cross_ref_request_date_range(keywords, date_from, date_to, **kwargs)

for json_response in json_responses:

message = json_response.get('message', {})
keyword = message.get('query').get('search-terms')
next_cursor = message.get('next-cursor')
total_results = message.get('total-results')
search_results = message.get('items', [{}]) if total_results > 0 else []

if final_result[keyword] is None:
final_result[keyword] = search_results
else:
final_result[keyword].extend(search_results)

if total_results > int(kwargs['rows']) and len(search_results) == int(kwargs['rows']):
kwargs['cursor'] = next_cursor
await self._send_client_request(final_result, [keyword], date_from, date_to, **kwargs)

async def cross_ref_request(self, request: CrossRefSearchRequest, **kwargs) -> dict[str, list[str]]:
date_from, date_to = _setup_date(request.date_from, request.date_to)
results: dict[str, list[str]] = dict.fromkeys(request.keywords)

await self._send_client_request(results, request.keywords, date_from, date_to, **kwargs)

return results


client.py

import asyncio
from aiohttp import ClientSession

async def _send_request_task(session: ClientSession, url: str):
try:
async with session.get(url) as response:
await response.read()
return response
# exception handler to come
except Exception as e:
print(f"exception for {url}")
print(str(e))


class CrossRefClient:
base_url = "https://api.crossref.org/works?" \
"query={}&" \
"filter=from-index-date:{},until-index-date:{}&" \
"sort={}&" \
"rows={}&" \
"cursor={}"

def __init__(self) -> None:
self.headers = {
"User-Agent": f"my_app/v0.1 (example.com/; mailto:youremail@domain.com) using FastAPI"
}

async def cross_ref_request_date_range(
self, keywords: [str], date_from: str, date_to: str, **kwargs
) -> list:
async with ClientSession(headers=self.headers) as session:
tasks = [
asyncio.create_task(
_send_request_task(session, self.base_url.format(
keyword, date_from, date_to, kwargs['sort'], kwargs['rows'], kwargs['cursor']
)),
name=TASK_NAME_BASE.format(keyword, date_from, date_to)
)
for keyword in keywords
]

responses = await asyncio.gather(*tasks)

return [await response.json() for response in responses]


How to optimize this better and use asynchronous calls better? Also this recursive loop might not be the best way to do it neither. Any ideas on that too?

I implemented a solution for synchronous calls and it's even slower. So I guess I'm not too far away.

Thanks!

Continue reading...