Styjun

I am trying to use aiohttp in one of my projects and struggling to figure out how to create a persistent aiohttp.ClientSession object. I have gone through the official aiohttp documentation but did not find it help in this context.

I have looked through other online forums and noticed that a lot has changed ever since aiohttp was created. In some examples on github, the aiohttp author is shown to be creating a ClientSession outside a coroutine functions (i.e. class Session: def __init__(self): self.session = aiohttp.ClientSession()). I also found that one should not create a ClientSession outside coroutine.

I have tried the following:

class Session:
 def __init__(self):
 self._session = None

 async def create_session(self):
 self._session = aiohttp.ClientSession()

 async fetch(self, url):
 if self._session is None:
 await self.create_session()
 async with self._session.get(url) as resp:
 return await resp.text()

I am getting a lot of warning about UnclosedSession and connector. I also frequently get SSLError. I also noticed that 2 out of three calls gets hung and I have to CTRL+C to kill it.

With requests I can simply initialize the session object in __init__, but it's not as simple as this with aiohttp.

I do not see any issues if I use the following (which is what I see as example all over the place) but unfortunately here I end up creating ClientSession with every request.

def fetch(url):
 async with aiohttp.ClientSession() as session:
 async with session.get(url) as resp:
 return await resp.text()

I can wrap aiohttp.ClientSession() in another function and use that as context-manager, but then too I would end up creating a new session object every time I call the wrapper function. I am trying to figure how to save a aiohttp.ClientSession in class namespace and reuse it.

Any help would be greatly appreciated.

Here is working example:

from aiohttp import ClientSession, TCPConnector
import asyncio


class CS:

 _cs: ClientSession

 def __init__(self):
 self._cs = ClientSession(connector=TCPConnector(verify_ssl=False))

 async def get(self, url):
 async with self._cs.get(url) as resp:
 return await resp.text()

 async def close(self):
 await self._cs.close()


async def func():
 cs = CS()
 print(await cs.get('https://google.com'))
 await cs.close() # you must close session


loop = asyncio.get_event_loop()
loop.run_until_complete(func())