validate db url in client ctor and in CLI (#355)

![Screenshot 2025-05-16 at 16 37
30](https://github.com/user-attachments/assets/fc67f67e-1dcb-4370-afd7-72cba6ba8201)

Author: maxdml

dbos/_client.py

@@ -3,8 +3,6 @@
 import uuid
 from typing import Any, Generic, List, Optional, TypedDict, TypeVar
 
-from sqlalchemy import URL
-
 from dbos._app_db import ApplicationDatabase
 from dbos._context import MaxPriority, MinPriority
 
@@ -15,6 +13,7 @@
 
 from dbos import _serialization
 from dbos._dbos import WorkflowHandle, WorkflowHandleAsync
+from dbos._dbos_config import is_valid_database_url
 from dbos._error import DBOSException, DBOSNonExistentWorkflowError
 from dbos._registrations import DEFAULT_MAX_RECOVERY_ATTEMPTS
 from dbos._serialization import WorkflowInputs
@@ -99,6 +98,7 @@ async def get_status(self) -> WorkflowStatus:
 
 class DBOSClient:
     def __init__(self, database_url: str, *, system_database: Optional[str] = None):
+        assert is_valid_database_url(database_url)
         # We only create database connections but do not run migrations
         self._sys_db = SystemDatabase(
             database_url=database_url,

dbos/_dbos_config.py

@@ -329,17 +329,9 @@ def process_config(
     if data.get("database_url") is not None and data["database_url"] != "":
         # Parse the db string and check required fields
         assert data["database_url"] is not None
+        assert is_valid_database_url(data["database_url"])
+
         url = make_url(data["database_url"])
-        required_fields = [
-            ("username", "Username must be specified in the connection URL"),
-            ("password", "Password must be specified in the connection URL"),
-            ("host", "Host must be specified in the connection URL"),
-            ("database", "Database name must be specified in the connection URL"),
-        ]
-        for field_name, error_message in required_fields:
-            field_value = getattr(url, field_name, None)
-            if not field_value:
-                raise DBOSInitializationError(error_message)
 
         if not data["database"].get("sys_db_name"):
             assert url.database is not None
@@ -385,6 +377,9 @@ def process_config(
     if not silent and logs["logLevel"] == "INFO" or logs["logLevel"] == "DEBUG":
         log_url = make_url(data["database_url"]).render_as_string(hide_password=True)
         print(f"[bold blue]Using database connection string: {log_url}[/bold blue]")
+        print(
+            f"[bold blue]Database engine parameters: {data['database']['db_engine_kwargs']}[/bold blue]"
+        )
 
     # Return data as ConfigFile type
     return data
@@ -432,6 +427,21 @@ def configure_db_engine_parameters(
     data["sys_db_engine_kwargs"] = system_engine_kwargs
 
 
+def is_valid_database_url(database_url: str) -> bool:
+    url = make_url(database_url)
+    required_fields = [
+        ("username", "Username must be specified in the connection URL"),
+        ("password", "Password must be specified in the connection URL"),
+        ("host", "Host must be specified in the connection URL"),
+        ("database", "Database name must be specified in the connection URL"),
+    ]
+    for field_name, error_message in required_fields:
+        field_value = getattr(url, field_name, None)
+        if not field_value:
+            raise DBOSInitializationError(error_message)
+    return True
+
+
 def _is_valid_app_name(name: str) -> bool:
     name_len = len(name)
     if name_len < 3 or name_len > 30:

dbos/cli/cli.py

@@ -18,7 +18,7 @@
 
 from .._app_db import ApplicationDatabase
 from .._client import DBOSClient
-from .._dbos_config import _is_valid_app_name, load_config
+from .._dbos_config import _is_valid_app_name, is_valid_database_url, load_config
 from .._docker_pg_helper import start_docker_pg, stop_docker_pg
 from .._schemas.system_database import SystemSchema
 from .._sys_db import SystemDatabase, reset_system_database
@@ -35,6 +35,7 @@ def _get_db_url(db_url: Optional[str]) -> str:
         raise ValueError(
             "Missing database URL: please set it using the --db-url flag or the DBOS_DATABASE_URL environment variable."
         )
+    assert is_valid_database_url(database_url)
     return database_url