b715755b76
Some checks failed
Build and Deploy / test-backend (push) Has been cancelled
Build and Deploy / build-backend (push) Has been cancelled
Build and Deploy / build-admin (push) Has been cancelled
Deploy to Production / build-backend (push) Has been cancelled
Deploy to Production / deploy (push) Has been cancelled
Co-authored-by: Cursor <cursoragent@cursor.com>
49 lines
1.7 KiB
Python
49 lines
1.7 KiB
Python
"""
|
||
统一配置加载：环境变量优先，fallback 到 .env，无 DATABASE_URL 时按 DB_* 拼接。
|
||
支持多环境（dev/prod）、容器（DB_HOST=db）、向后兼容。
|
||
"""
|
||
import os
|
||
from pathlib import Path
|
||
|
||
# 环境变量优先；fallback 到 .env（不覆盖已存在的系统变量）
|
||
def _load_dotenv():
|
||
try:
|
||
from dotenv import load_dotenv
|
||
except ImportError:
|
||
return
|
||
env = os.getenv("ENV", "").lower()
|
||
base = Path(__file__).resolve().parent.parent
|
||
if env == "prod":
|
||
load_dotenv(base / ".env.prod", override=False)
|
||
elif env == "dev":
|
||
load_dotenv(base / ".env.dev", override=False)
|
||
load_dotenv(base / ".env", override=False)
|
||
|
||
|
||
_load_dotenv()
|
||
|
||
|
||
def build_database_url_async() -> str:
|
||
"""拼接异步数据库 URL（postgresql+asyncpg）。"""
|
||
user = os.getenv("DB_USER", "wecom")
|
||
password = os.getenv("DB_PASSWORD", "password")
|
||
host = os.getenv("DB_HOST", "localhost")
|
||
port = os.getenv("DB_PORT", "5432")
|
||
db = os.getenv("DB_NAME", "wecom_ai")
|
||
return f"postgresql+asyncpg://{user}:{password}@{host}:{port}/{db}"
|
||
|
||
|
||
def build_database_url_sync() -> str:
|
||
"""拼接同步数据库 URL（postgresql，Alembic/psycopg2）。"""
|
||
user = os.getenv("DB_USER", "wecom")
|
||
password = os.getenv("DB_PASSWORD", "password")
|
||
host = os.getenv("DB_HOST", "localhost")
|
||
port = os.getenv("DB_PORT", "5432")
|
||
db = os.getenv("DB_NAME", "wecom_ai")
|
||
return f"postgresql://{user}:{password}@{host}:{port}/{db}"
|
||
|
||
|
||
# 优先使用环境变量，否则按 DB_* 拼接（向后兼容）
|
||
DATABASE_URL: str = os.getenv("DATABASE_URL") or build_database_url_async()
|
||
DATABASE_URL_SYNC: str = os.getenv("DATABASE_URL_SYNC") or build_database_url_sync()
|