65 lines
1.9 KiB
Python
65 lines
1.9 KiB
Python
from pathlib import Path
|
|
|
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
|
from pydantic import BaseModel, Field, field_validator, ValidationError, SecretStr
|
|
from sqlalchemy.engine import URL
|
|
|
|
BASE_DIR = Path(__file__).resolve().parents[2]
|
|
ENV_FILE_PATH = BASE_DIR / ".env"
|
|
|
|
SUPPORTED_DRIVERS = {
|
|
"postgresql": "postgresql+psycopg2",
|
|
"mysql": "mysql+pymysql",
|
|
"sqlserver": "mssql+pyodbc",
|
|
"sqlite": "sqlite"
|
|
}
|
|
|
|
class DatabaseConfig(BaseModel):
|
|
user: str
|
|
password: SecretStr
|
|
host: str
|
|
port: int
|
|
database: str
|
|
driver: str
|
|
|
|
@field_validator("driver")
|
|
def validate_driver(cls, v):
|
|
if v not in SUPPORTED_DRIVERS:
|
|
raise ValueError(f"Driver '{v}' não suportado. Use um de: {list(SUPPORTED_DRIVERS.keys())}")
|
|
return v
|
|
|
|
def sqlalchemy_url(self) -> URL:
|
|
drivername = SUPPORTED_DRIVERS[self.driver]
|
|
return URL.create(
|
|
drivername=drivername,
|
|
username=self.user,
|
|
password=self.password.get_secret_value(),
|
|
host=self.host,
|
|
port=self.port,
|
|
database=self.database,
|
|
)
|
|
|
|
class DatabaseSettings(BaseSettings):
|
|
database: dict[str, DatabaseConfig]
|
|
|
|
model_config = SettingsConfigDict(
|
|
env_file=ENV_FILE_PATH,
|
|
env_file_encoding="utf-8",
|
|
env_nested_delimiter="__",
|
|
extra="ignore"
|
|
)
|
|
|
|
class Settings:
|
|
databases = DatabaseSettings()
|
|
|
|
def get_url_database(self, database: str) -> URL:
|
|
try:
|
|
return self.databases.database[database].sqlalchemy_url()
|
|
except KeyError as e:
|
|
raise ValueError(f"Database '{database}' não encontrado. Use um de: {list(self.databases.database.keys())}")
|
|
|
|
def get_url_databases(self) -> dict[str, URL]:
|
|
return {
|
|
name: config.sqlalchemy_url() for name, config in self.databases.database.items()
|
|
}
|