修复数据库连接URL中密码包含"@"的问题

app/core/config.py

@@ -1,4 +1,5 @@
 from pathlib import Path
+from urllib.parse import quote_plus
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -60,12 +61,14 @@ class Settings(BaseSettings):
 
     @property
     def SQLALCHEMY_DATABASE_URI(self) -> str:
-        return f"postgresql://{self.DB_USER}:{self.DB_PASSWORD}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"
+        db_password = quote_plus(self.DB_PASSWORD)
+        return f"postgresql://{self.DB_USER}:{db_password}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"
 
     @property
     def METADATA_DATABASE_URI(self) -> str:
+        metadata_password = quote_plus(self.METADATA_DB_PASSWORD)
         return (
-            f"postgresql+psycopg://{self.METADATA_DB_USER}:{self.METADATA_DB_PASSWORD}"
+            f"postgresql+psycopg://{self.METADATA_DB_USER}:{metadata_password}"
             f"@{self.METADATA_DB_HOST}:{self.METADATA_DB_PORT}/{self.METADATA_DB_NAME}"
         )
 

app/infra/repositories/metadata_repository.py

@@ -5,10 +5,28 @@ from uuid import UUID
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.core.encryption import get_encryptor
+from app.core.encryption import get_encryptor, is_encryption_configured
 from app.infra.db.metadata import models
 
 
+def _normalize_postgres_dsn(dsn: str) -> str:
+    if not dsn or "://" not in dsn:
+        return dsn
+    scheme, rest = dsn.split("://", 1)
+    if scheme not in ("postgresql", "postgres", "postgresql+psycopg"):
+        return dsn
+    if "@" not in rest:
+        return dsn
+    userinfo, hostinfo = rest.rsplit("@", 1)
+    if ":" not in userinfo:
+        return dsn
+    username, password = userinfo.split(":", 1)
+    if "@" not in password:
+        return dsn
+    password = password.replace("@", "%40")
+    return f"{scheme}://{username}:{password}@{hostinfo}"
+
+
 @dataclass(frozen=True)
 class ProjectDbRouting:
     project_id: UUID
@@ -84,8 +102,12 @@ class MetadataRepository:
         record = result.scalar_one_or_none()
         if not record:
             return None
-        encryptor = get_encryptor()
+        if is_encryption_configured():
-        dsn = encryptor.decrypt(record.dsn_encrypted)
+            encryptor = get_encryptor()
+            dsn = encryptor.decrypt(record.dsn_encrypted)
+        else:
+            dsn = record.dsn_encrypted
+        dsn = _normalize_postgres_dsn(dsn)
         return ProjectDbRouting(
             project_id=record.project_id,
             db_role=record.db_role,
@@ -106,12 +128,14 @@ class MetadataRepository:
         record = result.scalar_one_or_none()
         if not record:
             return None
-        encryptor = get_encryptor()
+        if record.gs_admin_password_encrypted:
-        password = (
+            if is_encryption_configured():
-            encryptor.decrypt(record.gs_admin_password_encrypted)
+                encryptor = get_encryptor()
-            if record.gs_admin_password_encrypted
+                password = encryptor.decrypt(record.gs_admin_password_encrypted)
-            else None
+            else:
-        )
+                password = record.gs_admin_password_encrypted
+        else:
+            password = None
         return ProjectGeoServerInfo(
             project_id=record.project_id,
             gs_base_url=record.gs_base_url,