cfdaily
61 lines
1.7 KiB
Python
61 lines
1.7 KiB
Python
"""
|
|
数据库配置文件
|
|
支持 SQLite（方案零）和 PostgreSQL（方案一）
|
|
"""
|
|
import os
|
|
from typing import Optional
|
|
from pydantic_settings import BaseSettings
|
|
|
|
|
|
class DatabaseSettings(BaseSettings):
|
|
"""数据库配置"""
|
|
|
|
# 数据库类型: sqlite 或 postgresql
|
|
db_type: str = "sqlite"
|
|
|
|
# SQLite 配置
|
|
sqlite_path: str = os.path.join(os.path.dirname(__file__), "data", "quant_trading.db")
|
|
|
|
# PostgreSQL 配置（方案一使用）
|
|
postgres_host: str = "localhost"
|
|
postgres_port: int = 5432
|
|
postgres_user: str = "quant_user"
|
|
postgres_password: str = ""
|
|
postgres_db: str = "quant_trading"
|
|
|
|
# 连接池配置
|
|
pool_size: int = 5
|
|
max_overflow: int = 10
|
|
pool_timeout: int = 30
|
|
pool_recycle: int = 3600
|
|
|
|
# 日志配置
|
|
echo_sql: bool = False
|
|
|
|
class Config:
|
|
env_prefix = "QUANT_"
|
|
env_file = ".env"
|
|
|
|
def get_database_url(self) -> str:
|
|
"""获取数据库连接 URL"""
|
|
if self.db_type == "sqlite":
|
|
# 确保 SQLite 数据库目录存在
|
|
os.makedirs(os.path.dirname(self.sqlite_path), exist_ok=True)
|
|
return f"sqlite:///{self.sqlite_path}"
|
|
elif self.db_type == "postgresql":
|
|
return (
|
|
f"postgresql+psycopg2://{self.postgres_user}:{self.postgres_password}"
|
|
f"@{self.postgres_host}:{self.postgres_port}/{self.postgres_db}"
|
|
)
|
|
else:
|
|
raise ValueError(f"不支持的数据库类型: {self.db_type}")
|
|
|
|
|
|
# 全局数据库配置实例
|
|
db_settings = DatabaseSettings()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print(f"数据库类型: {db_settings.db_type}")
|
|
print(f"数据库连接 URL: {db_settings.get_database_url()}")
|