fix: broker shutdown before connection pool creation

Author: danfimov

src/taskiq_pg/aiopg/result_backend.py

@@ -76,8 +76,8 @@ async def startup(self) -> None:
 
     async def shutdown(self) -> None:
         """Close the connection pool."""
-        async with self._database_pool.acquire() as connection:
-            await connection.close()
+        if getattr(self, '_database_pool', None) is not None:
+            self._database_pool.close()
 
     async def set_result(
         self,

src/taskiq_pg/asyncpg/result_backend.py

@@ -6,7 +6,6 @@
 from taskiq import AsyncResultBackend, TaskiqResult
 from taskiq.compat import model_dump, model_validate
 from taskiq.serializers import PickleSerializer
-from typing_extensions import override
 
 from taskiq_pg.asyncpg.queries import (
     CREATE_INDEX_QUERY,
@@ -67,7 +66,6 @@ def dsn(self) -> str | None:
             return self._dsn()
         return self._dsn
 
-    @override
     async def startup(self) -> None:
         """
         Initialize the result backend.
@@ -96,12 +94,11 @@ async def startup(self) -> None:
             ),
         )
 
-    @override
     async def shutdown(self) -> None:
         """Close the connection pool."""
-        await self._database_pool.close()
+        if getattr(self, '_database_pool', None) is not None:
+            await self._database_pool.close()
 
-    @override
     async def set_result(
         self,
         task_id: str,
@@ -121,7 +118,6 @@ async def set_result(
             self.serializer.dumpb(model_dump(result)),
         )
 
-    @override
     async def is_result_ready(self, task_id: str) -> bool:
         """
         Returns whether the result is ready.
@@ -139,7 +135,6 @@ async def is_result_ready(self, task_id: str) -> bool:
             ),
         )
 
-    @override
     async def get_result(
         self,
         task_id: str,

src/taskiq_pg/psqlpy/result_backend.py

@@ -84,7 +84,8 @@ async def startup(self) -> None:
 
     async def shutdown(self) -> None:
         """Close the connection pool."""
-        self._database_pool.close()
+        if getattr(self, '_database_pool', None) is not None:
+            self._database_pool.close()
 
     async def set_result(
         self,