Hot fix (#37)

* adding env test

* fix: support DATABASE_URL with special chars and optional async deps

- Fix PGMQConfig URI parser to match libpq behavior (split on last @ in
  netloc) so passwords containing @, /, or + parse correctly
- Use re-assembled DSNs in queue.py/async_queue.py instead of passing
  malformed raw URIs to drivers
- Rewrite SQLAlchemy connection URLs to use explicit +psycopg/+asyncpg
  drivers so bare postgresql:// doesn't fall back to missing psycopg2
- Make async_queue import optional in __init__.py (ModuleNotFoundError
  when asyncpg is not installed)
- Add python-dotenv to dev deps and load .env in tests/__init__.py
- Update test utilities to respect DATABASE_URL and fall back to PG_*
  env vars for Supabase/remote testing

* add .env.bck

* fix close()

* bump the version

* update uv.lock

* fix: improve connection string parsing for DATABASE_URL

* fix: remove python-dotenv dependency from project

* fix: update connection string handling to use async_dsn for SQLAlchemy engines

Author: tavallaie

.gitignore

@@ -308,6 +308,7 @@ celerybeat.pid
 
 # Environments
 .env
+.env.bck
 .envrc
 .venv
 env/

Makefile

@@ -30,4 +30,7 @@ run-pgmq-postgres:
 
 test: clear-postgres run-pgmq-postgres
 	sleep 10  # Give PostgreSQL time to start
+	uv run python -m unittest discover -s tests -p "test_*.py"
+
+test-env:
 	uv run python -m unittest discover -s tests -p "test_*.py"

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "pgmq"
-version = "1.1.0"
+version = "1.1.1"
 description = "Python client for the PGMQ Postgres extension."
 readme = "README.md"
 license = "Apache-2.0"
@@ -25,7 +25,7 @@ classifiers = [
 
 authors = [
     { name = "Adam Hendel",email = "adam@hendel.dev" },
-    { name = "Ali Tavallaie", email = "ali@techbend.io" },
+    { name = "Ali Tavallaie", email = "ali@techbend.dev" },
 ]
 dependencies = [
     "orjson>=3.11.3",
@@ -48,6 +48,7 @@ bench = [
 dev = [
     "loguru>=0.7.3",
     "pre-commit>=4.3.0",
+    "python-dotenv>=1.2.2",
     "ruff>=0.12.12",
 ]
 docs = [

src/pgmq/__init__.py

@@ -19,14 +19,21 @@
 
 # Client classes (default psycopg-based)
 from pgmq.queue import PGMQueue as SyncPGMQueue
-from pgmq.async_queue import PGMQueue as AsyncPGMQueue
+
+try:
+    from pgmq.async_queue import PGMQueue as AsyncPGMQueue
+except ImportError:  # pragma: no cover
+    AsyncPGMQueue = None  # type: ignore[misc, assignment]
 
 # SQLAlchemy-based clients (available when sqlalchemy is installed)
 try:
     from pgmq.sqlalchemy_queue import PGMQueue as SQLAlchemyPGMQueue
-    from pgmq.sqlalchemy_async_queue import PGMQueue as SQLAlchemyAsyncPGMQueue
 except ImportError:  # pragma: no cover
     SQLAlchemyPGMQueue = None  # type: ignore[misc, assignment]
+
+try:
+    from pgmq.sqlalchemy_async_queue import PGMQueue as SQLAlchemyAsyncPGMQueue
+except ImportError:  # pragma: no cover
     SQLAlchemyAsyncPGMQueue = None  # type: ignore[misc, assignment]
 
 # Decorators

src/pgmq/async_queue.py

@@ -78,13 +78,10 @@ def __post_init__(self) -> None:
     async def init(self) -> None:
         """Initialize the asyncpg connection pool."""
         log_with_context(self.logger, logging.DEBUG, "Creating asyncpg pool")
-        dsn = (
-            self.config.conn_string
-            if self.config.conn_string
-            else self.config.async_dsn
-        )
+        # Always use the re-encoded URI so that malformed original URIs
+        # (e.g. passwords containing unescaped @ or /) are corrected.
         self.pool = await asyncpg.create_pool(
-            dsn,
+            self.config.async_dsn,
             min_size=1,
             max_size=self.config.pool_size,
         )

src/pgmq/base.py

@@ -70,23 +70,49 @@ def _parse_conn_string(self, conn_string: str) -> None:
         Supports:
         - URI: postgresql://user:pass@host:port/database
         - Libpq: host=localhost port=5432 dbname=database user=postgres password=postgres
+
+        For URIs, this uses a libpq-compatible parser that splits the netloc on
+        the *last* ``@`` (so passwords containing ``@`` work) and tolerates
+        unescaped ``/`` or ``+`` inside the password.
         """
         # URI Format
         if "://" in conn_string:
             try:
+                # Try standard urlparse first
                 parsed = urllib.parse.urlparse(conn_string)
 
-                if parsed.hostname:
-                    self.host = parsed.hostname
-                if parsed.port:
-                    self.port = str(parsed.port)
-                if parsed.path and len(parsed.path) > 1:
-                    # path is usually '/dbname', slice off the leading '/'
-                    self.database = parsed.path[1:]
-                if parsed.username:
-                    self.username = parsed.username
-                if parsed.password:
-                    self.password = parsed.password
+                # Mimic libpq: split on the LAST @ to handle passwords containing @
+                scheme, rest = conn_string.split("://", 1)
+                if "@" in rest:
+                    userinfo, hostpath = rest.rsplit("@", 1)
+                    if ":" in userinfo:
+                        self.username, self.password = userinfo.split(":", 1)
+                    else:
+                        self.username = userinfo
+
+                    self.username = urllib.parse.unquote(self.username)
+                    if self.password:
+                        self.password = urllib.parse.unquote(self.password)
+                else:
+                    hostpath = rest
+
+                # Parse hostpath (host:port/database?options)
+                # Prepend a dummy scheme to use urlparse safely on the remaining part
+                hp_parsed = urllib.parse.urlparse("http://" + hostpath)
+                if hp_parsed.hostname:
+                    self.host = hp_parsed.hostname
+                if hp_parsed.port:
+                    self.port = str(hp_parsed.port)
+                if hp_parsed.path and len(hp_parsed.path) > 1:
+                    # path is '/dbname', slice off the leading '/'
+                    self.database = urllib.parse.unquote(hp_parsed.path[1:])
+
+                # If urlparse gave us userinfo that we didn't extract above,
+                # prefer the rsplit result (handles passwords with @)
+                if not self.username and parsed.username is not None:
+                    self.username = urllib.parse.unquote(parsed.username)
+                if not self.password and parsed.password is not None:
+                    self.password = urllib.parse.unquote(parsed.password)
 
             except Exception as e:
                 raise ValueError(f"Failed to parse connection URI: {e}")

src/pgmq/queue.py

@@ -71,9 +71,10 @@ def __post_init__(self):
     def _init_pool(self) -> None:
         """Initialize the connection pool."""
         log_with_context(self.logger, logging.DEBUG, "Creating connection pool")
-        dsn = self.config.conn_string if self.config.conn_string else self.config.dsn
+        # Always use the re-assembled libpq DSN so that malformed URIs
+        # (e.g. passwords containing unescaped @ or /) are corrected.
         self.pool = ConnectionPool(
-            dsn,
+            self.config.dsn,
             min_size=1,
             max_size=self.config.pool_size,
             open=True,

src/pgmq/sqlalchemy_async_queue.py

@@ -11,7 +11,6 @@
 from datetime import datetime
 import os
 import logging
-import urllib.parse
 from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine, async_sessionmaker
 from sqlalchemy.pool import AsyncAdaptedQueuePool
 from sqlalchemy import text
@@ -97,17 +96,11 @@ async def init(self) -> None:
             return
 
         log_with_context(self.logger, logging.DEBUG, "Creating async SQLAlchemy engine")
-        if self.config.conn_string:
-            # If a full connection string is provided, use it directly
-            connection_url = self.config.conn_string
-        else:
-            # Otherwise, construct it from individual components
-            user = urllib.parse.quote_plus(self.config.username)
-            password = urllib.parse.quote_plus(self.config.password)
-            connection_url = (
-                f"postgresql+asyncpg://{user}:{password}@"
-                f"{self.config.host}:{self.config.port}/{self.config.database}"
-            )
+        # Use the re-assembled and quoted URI from config, swapping the driver prefix.
+        # This handles both URI and libpq input formats and fixes malformed credentials.
+        connection_url = self.config.async_dsn.replace(
+            "postgresql://", "postgresql+asyncpg://", 1
+        )
         self.engine = create_async_engine(
             connection_url,
             poolclass=AsyncAdaptedQueuePool,

src/pgmq/sqlalchemy_queue.py

@@ -12,7 +12,6 @@
 from datetime import datetime
 import os
 import logging
-import urllib.parse
 import warnings
 from sqlalchemy import create_engine, text, Engine
 from sqlalchemy.orm import sessionmaker, Session
@@ -92,17 +91,11 @@ def __post_init__(self):
     def _init_engine(self) -> None:
         """Initialize the SQLAlchemy engine."""
         log_with_context(self.logger, logging.DEBUG, "Creating SQLAlchemy engine")
-        if self.config.conn_string:
-            # If a full connection string is provided, use it directly
-            connection_url = self.config.conn_string
-        else:
-            # Otherwise, construct it from individual components
-            user = urllib.parse.quote_plus(self.config.username)
-            password = urllib.parse.quote_plus(self.config.password)
-            connection_url = (
-                f"postgresql+psycopg://{user}:{password}@"
-                f"{self.config.host}:{self.config.port}/{self.config.database}"
-            )
+        # Use the re-assembled and quoted URI from config, swapping the driver prefix.
+        # This handles both URI and libpq input formats and fixes malformed credentials.
+        connection_url = self.config.async_dsn.replace(
+            "postgresql://", "postgresql+psycopg://", 1
+        )
         self.engine = create_engine(
             connection_url,
             poolclass=QueuePool,

tests/__init__.py

@@ -0,0 +1,5 @@
+# Automatically load .env file so DATABASE_URL / PG_* variables are available
+# to PGMQConfig even when tests are run without `uv run --env-file`.
+from dotenv import load_dotenv
+
+load_dotenv()