Files
dababase-etl-python/config.py
2026-03-04 12:17:52 +08:00

61 lines
2.1 KiB
Python

import os
from typing import Dict, Any
from dotenv import load_dotenv
# 加载环境变量文件
load_dotenv()
class DatabaseConfig:
"""数据库配置类"""
# MySQL配置
MYSQL_CONFIG = {
"host": os.getenv("MYSQL_HOST", "localhost"),
"port": int(os.getenv("MYSQL_PORT", "3306")),
"username": os.getenv("MYSQL_USERNAME", "root"),
"password": os.getenv("MYSQL_PASSWORD", "password"),
"database": os.getenv("MYSQL_DATABASE", "test_db")
}
# Oracle配置
ORACLE_CONFIG = {
"host": os.getenv("ORACLE_HOST", "192.168.13.27"),
"port": int(os.getenv("ORACLE_PORT", "1521")),
"username": os.getenv("ORACLE_USERNAME", "bizuser"),
"password": os.getenv("ORACLE_PASSWORD", "MySecurePass123"),
"service_name": os.getenv("ORACLE_SERVICE_NAME", "ORCLPDB1")
}
# SQL Server配置
SQLSERVER_CONFIG = {
"host": os.getenv("SQLSERVER_HOST", "localhost"),
"port": int(os.getenv("SQLSERVER_PORT", "1433")),
"username": os.getenv("SQLSERVER_USERNAME", "sa"),
"password": os.getenv("SQLSERVER_PASSWORD", "password"),
"database": os.getenv("SQLSERVER_DATABASE", "master")
}
# PostgreSQL配置
POSTGRESQL_CONFIG = {
"host": os.getenv("POSTGRESQL_HOST", "localhost"),
"port": int(os.getenv("POSTGRESQL_PORT", "5432")),
"username": os.getenv("POSTGRESQL_USERNAME", "postgres"),
"password": os.getenv("POSTGRESQL_PASSWORD", "password"),
"database": os.getenv("POSTGRESQL_DATABASE", "postgres")
}
@classmethod
def get_config(cls, db_type: str) -> Dict[str, Any]:
"""根据数据库类型获取配置"""
config_map = {
"mysql": cls.MYSQL_CONFIG,
"oracle": cls.ORACLE_CONFIG,
"sqlserver": cls.SQLSERVER_CONFIG,
"postgresql": cls.POSTGRESQL_CONFIG
}
return config_map.get(db_type.lower(), {})
@classmethod
def is_sample_data_enabled(cls) -> bool:
"""检查是否启用示例数据初始化"""
return os.getenv("ENABLE_SAMPLE_DATA", "true").lower() == "true"