Allow password as a callable when using Postgres backend (#250)

* fix: Ensure password is overrideable

Removes the explicit naming of the password parameter for asyncpg, and
moves the setting of the password to the kwargs. Further adds tests to
ensure that the password argument is preserved.

* fix: Add typing for kwargs dict

databases/backends/postgres.py

@@ -44,7 +44,7 @@ def _get_dialect(self) -> Dialect:
     def _get_connection_kwargs(self) -> dict:
         url_options = self._database_url.options
 
-        kwargs = {}
+        kwargs = {}  # type: typing.Dict[str, typing.Any]
         min_size = url_options.get("min_size")
         max_size = url_options.get("max_size")
         ssl = url_options.get("ssl")
@@ -62,15 +62,15 @@ def _get_connection_kwargs(self) -> dict:
 
     async def connect(self) -> None:
         assert self._pool is None, "DatabaseBackend is already running"
-        kwargs = self._get_connection_kwargs()
-        self._pool = await asyncpg.create_pool(
+        kwargs = dict(
             host=self._database_url.hostname,
             port=self._database_url.port,
             user=self._database_url.username,
             password=self._database_url.password,
             database=self._database_url.database,
-            **kwargs,
         )
+        kwargs.update(self._get_connection_kwargs())
+        self._pool = await asyncpg.create_pool(**kwargs)
 
     async def disconnect(self) -> None:
         assert self._pool is not None, "DatabaseBackend is not running"

tests/test_connection_options.py

@@ -43,6 +43,24 @@ def test_postgres_explicit_ssl():
     assert kwargs == {"ssl": True}
 
 
+def test_postgres_no_extra_options():
+    backend = PostgresBackend("postgres://localhost/database")
+    kwargs = backend._get_connection_kwargs()
+    assert kwargs == {}
+
+
+def test_postgres_password_as_callable():
+    def gen_password():
+        return "Foo"
+
+    backend = PostgresBackend(
+        "postgres://:password@localhost/database", password=gen_password
+    )
+    kwargs = backend._get_connection_kwargs()
+    assert kwargs == {"password": gen_password}
+    assert kwargs["password"]() == "Foo"
+
+
 def test_mysql_pool_size():
     backend = MySQLBackend("mysql://localhost/database?min_size=1&max_size=20")
     kwargs = backend._get_connection_kwargs()