chore: encrypt_secret.py

migrations/env.py

@@ -7,6 +7,7 @@ from logging.config import fileConfig
 from typing import Any
 
 from alembic import context
+from common.config import _decrypt_value, settings
 from common.db import Base
 from sqlalchemy import Connection, pool
 from sqlalchemy.ext.asyncio import async_engine_from_config
@@ -26,9 +27,7 @@ def canonical_default(value: Any) -> tuple[str, Any] | None:
 
     text_value = str(value).strip()
     while (
-        len(text_value) >= 2
+        len(text_value) >= 2 and text_value.startswith("(") and text_value.endswith(")")
-        and text_value.startswith("(")
-        and text_value.endswith(")")
     ):
         text_value = text_value[1:-1].strip()
     if (
@@ -62,13 +61,22 @@ def compare_server_default(
 
 
 def database_url() -> str:
-    """Return the runtime database URL without storing credentials in the repo."""
+    """Return the runtime database URL and auto-decrypt ENC(...) if present."""
-    try:
+    # 优先使用 settings (Pydantic 已拦截并解密)
-        return os.environ["DATABASE_URL"]
+    url = getattr(settings, "database_url", None)
-    except KeyError as exc:
+
-        raise RuntimeError(
+    # 回退机制：如果通过环境变量直接传入且未被 settings 解析
-            "DATABASE_URL is required for Alembic commands"
+    if not url:
-        ) from exc
+        url = os.getenv("DATABASE_URL")
+
+    if not url:
+        raise RuntimeError("DATABASE_URL is required for Alembic commands")
+
+    # 如果变量值仍然包含 ENC(...) 前缀（例如直接拿到的环境变量），手动解密
+    if url.startswith("ENC("):
+        url = _decrypt_value(url)
+
+    return url
 
 
 def configure_context(*, connection: Connection | None = None) -> None: