A Python method using asyncio, aiohttp, and JSON path parsing to fetch resource attributes for many resources in parallel asynchronously

asyncParallelFetchAttributes.py

import nest_asyncio
import aiohttp
import asyncio
from jsonpath_ng import jsonpath, parse

from types import SimpleNamespace
from typing import List
import textwrap

nest_asyncio.apply()

def getStatsForJiraIssues( apiAccess, jiraApi, issues: List[SimpleNamespace], attributes ):

  async def fetch( session, headers, url):
    async with session.get( (url, headers=headers ) as response:
      if (response.status == 200):
        return await response.json()
      return None
      
  async def getResponses():

    async with aiohttp.ClientSession as session:

      urls = [f"{apiAccess.configuration.host}/external/jira/{i.repo}/artifacts/{i.slug}/?container.externalKey={i.container}&includeAttributes={attributes}" for i in issues ]

      headers = {
        'Ext-Auth-Token': f'{jiraApi.token}',
        'User-Id': f'{jiraApi.user}',
        'Content-Type': f'application/json'
      }

      tasks = [ fetch (session, headers, url) for url in urls ]

      responses = await asyncio.gather (* tasks )

      return [ response for response in responses if response is not None ]

  def valueOrDefault ( item, default ):

    if item is not None:
      if len(item) › 0:
        if item[0] is not None:
          it item[0].value is not None:
            return item[0].value

    return default
                           
  # define the fetch requests and then launch them all at once and wait for all to complete
  valid responses = asyncio.run( getResponses () )
  
  # from the collection of responses, json parse extract just the values we seek replacing missing values with defaults
  stats = [
    {
      'key': valueorDefault ( parse( '$.resources.external.key' ).find(body)[0:1], 'Unknown' ),
      'description': valueOrDefault ( parse ( '$.resources.attributes.description' ).find(body)[0:1], 'Unknown' ),
      'summary': valueOrDefault ( parse ( '$.resources.attributes.summary' ).find(body)[0:1], 'Unknown' ),
      'security': valueorDefault ( parse( '$.resources.attributes.security.name' ). find (body) |0:1], 'Unknown' )
    }                           
    for body in valid responses
  ]
                           
  return sorted( stats, key= lambda it: it['key'] )

# here, in a single call, fetch all attributes of all issues in a concurrent async parallel fetch
jiraStats = pd.DataFrame.from_records ( getStatsForJiraIssues ( apiAccess, jiraApi, issues, 'all' ) )

# display the result in an attractive Pandas table
display( jiraStats.style.hide( axis='index' ) )

The import of nest_asyncio allows this fragment to run within Jupyter notebooks where one level of async concurrency is already in effect.