Merge pull request #211 from euri10/scheme_url

raise an error if from_url argument misses the scheme

saq/errors.py

@@ -5,3 +5,7 @@
 
 class MissingDependencyError(Exception):
     pass
+
+
+class InvalidUrlError(Exception):
+    pass

saq/queue/base.py

@@ -10,7 +10,9 @@
 import typing as t
 from abc import ABC, abstractmethod
 from contextlib import asynccontextmanager
+from urllib.parse import urlparse
 
+from saq.errors import InvalidUrlError
 from saq.job import (
     TERMINAL_STATUSES,
     UNSUCCESSFUL_TERMINAL_STATUSES,
@@ -172,20 +174,23 @@ async def _enqueue(self, job: Job) -> Job | None:
 
     @staticmethod
     def from_url(url: str, **kwargs: t.Any) -> Queue:
-        """Create a queue with a Postgers or Redis url."""
-        if url.startswith("redis"):
+        """Create a queue with either a redis, postgres or http url."""
+        parsed_url = urlparse(url)
+        scheme = parsed_url.scheme.lower()
+        if scheme.startswith("redis"):
             from saq.queue.redis import RedisQueue
 
             return RedisQueue.from_url(url, **kwargs)
-
-        if url.startswith("postgres"):
+        elif scheme.startswith("postgres"):
             from saq.queue.postgres import PostgresQueue
 
             return PostgresQueue.from_url(url, **kwargs)
+        elif scheme.startswith("http"):
+            from saq.queue.http import HttpQueue
 
-        from saq.queue.http import HttpQueue
-
-        return HttpQueue.from_url(url, **kwargs)
+            return HttpQueue.from_url(url, **kwargs)
+        else:
+            raise InvalidUrlError(f"Invalid url: {url}")
 
     async def connect(self) -> None:
         self._loop = asyncio.get_running_loop()

tests/helpers.py

@@ -10,8 +10,8 @@
 POSTGRES_TEST_SCHEMA = "test_saq"
 
 
-async def create_redis_queue(**kwargs: t.Any) -> RedisQueue:
-    queue = t.cast(RedisQueue, Queue.from_url("redis://localhost:6379", **kwargs))
+async def create_redis_queue(url="redis://localhost:6379", **kwargs: t.Any) -> RedisQueue:
+    queue = t.cast(RedisQueue, Queue.from_url(url, **kwargs))
     await queue.connect()
     await queue.redis.flushdb()
     return queue

tests/test_queue.py

@@ -6,10 +6,12 @@
 import time
 import typing as t
 import unittest
+from functools import partial
 from unittest import mock
 
 from psycopg.sql import SQL
 
+from saq.errors import InvalidUrlError
 from saq.job import Job, Status
 from saq.queue import JobError, Queue
 from saq.utils import uuid1
@@ -42,6 +44,18 @@ async def error(_ctx: Context) -> None:
 functions: list[Function] = [echo, error]
 
 
+class TestQueueError(unittest.IsolatedAsyncioTestCase):
+    queue: Queue
+    create_queue: t.Callable
+
+    async def asyncSetUp(self) -> None:
+        self.create_queue = partial(create_redis_queue, "scheme://localhost")
+
+    async def test_queue(self) -> None:
+        with self.assertRaises(InvalidUrlError):
+            self.queue: RedisQueue = await self.create_queue()
+
+
 class TestQueue(unittest.IsolatedAsyncioTestCase):
     queue: Queue
     create_queue: t.Callable

tests/test_worker.py

@@ -659,7 +659,7 @@ async def asyncSetUp(self) -> None:
         await super().asyncSetUp()
 
         async def create_http_queue():
-            queue = Queue.from_url("")
+            queue = Queue.from_url("http://")
             queue.session = self.client
             return queue