Merge pull request #184 from tobymao/feat-store-event-loop-in-a-queue

Feat: Preserve the event loop in a Queue when connecting

saq/queue/base.py

@@ -68,10 +68,15 @@ def __init__(
         self._dump = dump or json.dumps
         self._load = load or json.loads
         self._before_enqueues: dict[int, BeforeEnqueueType] = {}
+        self._loop: asyncio.AbstractEventLoop | None = None
 
     def job_id(self, job_key: str) -> str:
         return job_key
 
+    @property
+    def loop(self) -> asyncio.AbstractEventLoop:
+        return self._loop or asyncio.get_running_loop()
+
     @abstractmethod
     async def disconnect(self) -> None:
         pass
@@ -167,7 +172,7 @@ def from_url(url: str, **kwargs: t.Any) -> Queue:
         return HttpQueue.from_url(url, **kwargs)
 
     async def connect(self) -> None:
-        pass
+        self._loop = asyncio.get_running_loop()
 
     def serialize(self, job: Job) -> bytes | str:
         return self._dump(job.to_dict())

saq/queue/http.py

@@ -135,6 +135,7 @@ def __init__(
     async def connect(self) -> None:
         if not self.session:
             self.session = ClientSession(**self.session_kwargs)
+        await super().connect()
 
     async def disconnect(self) -> None:
         if self.session:

saq/queue/postgres.py

@@ -145,6 +145,8 @@ async def connect(self) -> None:
         await self.pool.resize(min_size=self.min_size, max_size=self.max_size)
         await self.init_db()
 
+        await super().connect()
+
     def serialize(self, job: Job) -> bytes | str:
         """Ensure serialized job is in bytes because the job column is of type BYTEA."""
         serialized = self._dump(job.to_dict())