Skip to content

Instantly share code, notes, and snippets.

@Auax
Created April 24, 2022 19:27
Show Gist options
  • Save Auax/f6b5feec3f4c8365303dc8a54e57d23e to your computer and use it in GitHub Desktop.
Save Auax/f6b5feec3f4c8365303dc8a54e57d23e to your computer and use it in GitHub Desktop.
Base File to send AIOHTTP Asynchronus requests using Python 3
import asyncio
import json
import time
from pystyle import Colors, Colorate
from typing import Dict, Any, List, Tuple
from aiohttp import ClientSession
from yarl import URL
class AsyncAIOHTTP:
def __init__(self, session: ClientSession):
self.session = session
async def http_get_with_aiohttp(self,
url: str | URL,
headers: Dict = {},
proxy: str = None,
timeout: int = 10,
verbose: bool = False) -> (int, Dict[str, Any], bytes):
"""
Perform an asynchronus aiohttp request
:param url:
:param headers:
:param proxy:
:param timeout:
:param verbose: print output
:return: response.status [int], response_json [Dict[str, Any]], response_content [bytes]
"""
response = await self.session.get(url=url, headers=headers, proxy=proxy, timeout=timeout)
response_json = None
try:
response_json = await response.json(content_type=None)
except json.decoder.JSONDecodeError as E:
if verbose:
print(Colorate.Color(Colors.red, "Error: ") + E)
response_content = None
try:
response_content = await response.read()
except Exception as E:
if verbose:
print(Colorate.Color(Colors.red, "Error: ") + E)
if verbose:
print(Colorate.Color(Colors.green, "Status: ") + str(response.status))
return response.status, response_json, response_content
async def http_get_with_aiohttp_parallel(self,
list_of_urls: List[str],
headers: Dict = {},
proxy: str = None,
timeout: int = 10,
verbose: bool = False) -> (List[Tuple[int, Dict[str, Any], bytes]], float):
"""
Execute several asynchronus aiohttp requests from a URL list
:param list_of_urls:
:param headers:
:param proxy: use a custom proxy
:param timeout: timeout of the requests
:param verbose: print output
:return: results [List[Tuple[int, Dict[str, Any], bytes]]], t [float]
"""
t1 = time.time()
results = await asyncio.gather(
*[self.http_get_with_aiohttp(
url,
headers,
proxy,
timeout,
verbose
)
for url in list_of_urls]
)
t2 = time.time()
t = t2 - t1
return results, t
async def close_session(self):
await self.session.close()
@staticmethod
def run(func: Any):
asyncio.run(func())
async def to_run_shares():
session = ClientSession()
shareAsync = AsyncAIOHTTP(session)
# URL list
urls = ["https://api.myip.com/" for _ in range(0, 1000)]
results, t = await shareAsync.http_get_with_aiohttp_parallel(
urls,
verbose=True
)
print(Colorate.Color(Colors.gray, f"Done in {t:.2f}s!"))
await shareAsync.close_session()
# Fix Windows RuntimeError
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
# Run
asyncio.run(to_run_shares())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment