fix unclosed asyncio session error

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "upoly"
-version = "0.1.33"
+version = "0.1.34"
 description = "High performance asyncio REST client for polygon.io"
 authors = ["Riley <[email protected]>"]
 license = "AGPL"
@@ -26,13 +26,13 @@ classifiers = [
 [tool.poetry.dependencies]
 python = ">=3.8,<3.9"
 python-dotenv = "^0.15.0"
-orjson = "^3.4.6"
+orjson = "^3.5.0"
 nest-asyncio = "^1.4.3"
 pandas-market-calendars = "^1.6.1"
 joblib = "^1.0.1"
 numpy = "^1.20.1"
 pandas = "^1.2.2"
-aiohttp = "^3.7.3"
+aiohttp = "^3.7.4"
 aiodns = "^2.0.0"
 Brotli = "^1.0.9"
 cchardet = "^2.1.7"

tests/test_upoly.py

@@ -12,7 +12,7 @@
 
 
 def test_version() -> None:
-    assert __version__ == "0.1.33"
+    assert __version__ == "0.1.34"
 
 
 def test_create_dataset_from_polygon():

upoly/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "0.1.33"
+__version__ = "0.1.34"
 
 from .polygon_plus import NY as NY
 from .polygon_plus import async_polygon_aggs as async_polygon_aggs

upoly/polygon_plus.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 import os
 import sys
 from asyncio.proactor_events import _ProactorBasePipeTransport
@@ -19,6 +20,7 @@
 import pandas as pd
 import pandas_market_calendars as mcal
 import pytz
+from dotenv import find_dotenv, load_dotenv
 
 # import uvloop
 from joblib import Memory
@@ -35,6 +37,14 @@
 
 NY = pytz.timezone("America/New_York")
 
+try:
+    load_dotenv(find_dotenv(raise_error_if_not_found=True))
+except IOError:
+    try:
+        load_dotenv(find_dotenv(raise_error_if_not_found=True, usecwd=True))
+    except IOError:
+        logging.warning("missing .env file")
+
 
 # prevent errors in Jupyter/Ipython; otherwise attempt to use enhanced event loop
 try:
@@ -60,7 +70,11 @@ def typed_cache(
     fn: F,
 ) -> F:
     def wrapper(*args, **kwargs):
-        return memory.cache(fn)(*args, **kwargs)
+        is_debug = os.environ.get("UPOLY_DEBUG", None)
+        if is_debug:
+            return fn(*args, **kwargs)
+        else:
+            return memory.cache(fn)(*args, **kwargs)
 
     return cast(F, wrapper)
 
@@ -136,33 +150,36 @@ async def _dispatch_consume_polygon(
 
     queue: asyncio.Queue[bytes] = asyncio.Queue()
 
-    POLYGON_KEY_ID = unwrap(os.getenv("POLYGON_KEY_ID"))
+    try:
+        POLYGON_KEY_ID = unwrap(os.getenv("POLYGON_KEY_ID"))
+    except ValueError:
+        raise ValueError(f"Missing POLYGON_KEY_ID, please load with from python-dotenv")
 
     client = aiohttp.ClientSession()
-
-    await asyncio.gather(
-        *(
-            _produce_polygon_aggs(
-                POLYGON_KEY_ID,
-                client,
-                queue,
-                symbol,
-                timespan,
-                interval,
-                _from,
-                to,
-                unadjusted,
+    async with client:
+
+        await asyncio.gather(
+            *(
+                _produce_polygon_aggs(
+                    POLYGON_KEY_ID,
+                    client,
+                    queue,
+                    symbol,
+                    timespan,
+                    interval,
+                    _from,
+                    to,
+                    unadjusted,
+                )
+                for _from, to in intervals
             )
-            for _from, to in intervals
         )
-    )
-    await client.close()
-    results: List[bytes] = []
-    while not queue.empty():
-        results.append(await queue.get())
-        queue.task_done()
+        results: List[bytes] = []
+        while not queue.empty():
+            results.append(await queue.get())
+            queue.task_done()
 
-    return results
+        return results
 
 
 def combine_chunks(chunks: List[bytes]) -> Iterator[PolyAggResponse]: