async

Author: Carreau

private-sec-reporting.py

@@ -1,25 +1,26 @@
 import os
-import requests
-import os
 import json
-org = "jupyter"
+import asyncio
+import aiohttp
+from rich import print
 
+org = "jupyter"
 token = os.getenv("GH_TOKEN")
 
-def check_private_vulnerability_reporting(repo_name):
+async def check_private_vulnerability_reporting(session, repo_name):
     headers = {
         'Authorization': f'token {token}',
         'Accept': 'application/vnd.github.v3+json'
     }
     url = f'https://api.github.com/repos/{org}/{repo_name}/private-vulnerability-reporting'
 
-    response = requests.get(url, headers=headers)
-    
-    if response.status_code == 200:
-        return response.json().get('enabled', False)
+    async with session.get(url, headers=headers) as response:
+        if response.status == 200:
+            data = await response.json()
+            return data.get('enabled', False)
     return False
 
-def get_org_repos():
+async def get_org_repos(session):
     headers = {
         'Authorization': f'token {token}',
         'Accept': 'application/vnd.github.v3+json'
@@ -29,31 +30,38 @@ def get_org_repos():
     page = 1
     while True:
         url = f'https://api.github.com/orgs/{org}/repos?page={page}&per_page=100'
-        response = requests.get(url, headers=headers)
-        
-        if response.status_code != 200:
-            print(f"Error fetching repos: {response.status_code}")
-            break
-            
-        page_repos = response.json()
-        if not page_repos:
-            break
-         
-        repos.extend(page_repos)
-        page += 1
+        async with session.get(url, headers=headers) as response:
+            if response.status != 200:
+                print(f"Error fetching repos: {response.status}")
+                break
+                
+            page_repos = await response.json()
+            if not page_repos:
+                break
+             
+            repos.extend(page_repos)
+            page += 1
 
     return repos
 
-# Get all repos and check their vulnerability reporting status
-repos = get_org_repos()
-results = {}
-from rich import print
-for repo in repos:
-    repo_name = repo['name']
-    repo_is_private = repo['private']
-    if repo_is_private:
-        print(f"{repo_name:>25}: [yellow]Private[/yellow]")
-        continue
-    has_vuln_reporting = check_private_vulnerability_reporting(repo_name)
-    results[repo_name] = has_vuln_reporting
-    print(f"{repo_name:>25}: {'[green]Enabled[/green]' if has_vuln_reporting else '[red]Disabled[/red]'}")
+async def main():
+    async with aiohttp.ClientSession() as session:
+        repos = await get_org_repos(session)
+        tasks = []
+        
+        for repo in repos:
+            repo_name = repo['name']
+            if repo['private']:
+                print(f"{repo_name:>25}: [yellow]Private[/yellow]")
+                continue
+            
+            task = check_private_vulnerability_reporting(session, repo_name)
+            tasks.append((repo_name, task))
+        
+        results = await asyncio.gather(*[task for _, task in tasks])
+        
+        for (repo_name, _), has_vuln_reporting in zip(tasks, results):
+            print(f"{repo_name:>25}: {'[green]Enabled[/green]' if has_vuln_reporting else '[red]Disabled[/red]'}")
+
+if __name__ == "__main__":
+    asyncio.run(main())