Fix pg_dump backup auth and servers API bearer timeouts.

db/backup.py

@@ -3,15 +3,23 @@
 from __future__ import annotations
 
 import logging
+import os
 import subprocess
 from datetime import datetime, timezone
 
-from .connection import get_database_url
+from .connection import get_pg_connection_params
 from .store import invalidate_data_cache
 
 logger = logging.getLogger(__name__)
 
 
+def _pg_cli_env(password: str) -> dict[str, str]:
+    env = os.environ.copy()
+    if password:
+        env['PGPASSWORD'] = password
+    return env
+
+
 def backup_filename() -> str:
     stamp = datetime.now(timezone.utc).strftime('%Y-%m-%d_%H%M%S')
     return f'amnezia_panel_backup_{stamp}.sql'
@@ -19,11 +27,14 @@ def backup_filename() -> str:
 
 def export_database_sql() -> bytes:
     """Create a plain SQL dump of the panel PostgreSQL database."""
-    url = get_database_url()
+    params = get_pg_connection_params()
     proc = subprocess.run(
         [
             'pg_dump',
-            '--dbname', url,
+            '-h', params['host'],
+            '-p', params['port'],
+            '-U', params['user'],
+            '-d', params['dbname'],
             '--no-owner',
             '--no-acl',
             '--clean',
@@ -31,6 +42,7 @@ def export_database_sql() -> bytes:
         ],
         capture_output=True,
         check=False,
+        env=_pg_cli_env(params['password']),
     )
     if proc.returncode != 0:
         err = proc.stderr.decode('utf-8', errors='replace').strip()
@@ -44,12 +56,21 @@ def restore_database_sql(data: bytes) -> None:
     """Restore panel data from a plain SQL dump produced by pg_dump."""
     if not data or not data.strip():
         raise ValueError('Empty backup file')
-    url = get_database_url()
+    params = get_pg_connection_params()
     proc = subprocess.run(
-        ['psql', '--dbname', url, '-v', 'ON_ERROR_STOP=1', '-q'],
+        [
+            'psql',
+            '-h', params['host'],
+            '-p', params['port'],
+            '-U', params['user'],
+            '-d', params['dbname'],
+            '-v', 'ON_ERROR_STOP=1',
+            '-q',
+        ],
         input=data,
         capture_output=True,
         check=False,
+        env=_pg_cli_env(params['password']),
     )
     if proc.returncode != 0:
         err = proc.stderr.decode('utf-8', errors='replace').strip()

db/connection.py

@@ -24,6 +24,37 @@ def get_database_url() -> str:
     return os.environ.get('DATABASE_URL', DEFAULT_DATABASE_URL).strip()
 
 
+def get_pg_connection_params() -> dict[str, str]:
+    """Connection parameters for pg_dump/psql (same source as the app pool)."""
+    from psycopg.conninfo import conninfo_to_dict
+
+    url = get_database_url()
+    scheme, _, rest = url.partition('://')
+    if scheme.startswith('postgresql'):
+        url = f'postgresql://{rest}'
+
+    info = conninfo_to_dict(url)
+    params = {
+        'host': str(info.get('host') or 'localhost'),
+        'port': str(info.get('port') or '5432'),
+        'user': str(info.get('user') or 'amnezia'),
+        'password': str(info.get('password') or ''),
+        'dbname': str(info.get('dbname') or 'amnezia_panel'),
+    }
+
+    # Prefer discrete env vars when set (Dokploy / compose); avoids URL encoding issues.
+    if os.environ.get('POSTGRES_USER', '').strip():
+        params['user'] = os.environ['POSTGRES_USER'].strip()
+    if os.environ.get('POSTGRES_PASSWORD', '').strip():
+        params['password'] = os.environ['POSTGRES_PASSWORD'].strip()
+    if os.environ.get('POSTGRES_DB', '').strip():
+        params['dbname'] = os.environ['POSTGRES_DB'].strip()
+    if os.environ.get('POSTGRES_PORT', '').strip():
+        params['port'] = os.environ['POSTGRES_PORT'].strip()
+
+    return params
+
+
 def get_pool():
     global _pool
     if _pool is not None: