243 lines
7.6 KiB
Python
243 lines
7.6 KiB
Python
from __future__ import annotations
|
|
import os
|
|
import os
|
|
from typing import Dict, Any, Optional, Type
|
|
from urllib.parse import quote_plus
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from dotenv import load_dotenv
|
|
except Exception:
|
|
load_dotenv = None
|
|
|
|
def _load_dotenv_if_present() -> None:
|
|
"""
|
|
Load .env once if present. Priority rules:
|
|
1) CRUDKIT_DOTENV points to a file
|
|
2) Project root's .env (two dirs up from this file)
|
|
3) Current working directory .env
|
|
|
|
Env already present in the process takes precedence.
|
|
"""
|
|
if load_dotenv is None:
|
|
return
|
|
|
|
path_hint = os.getenv("CRUDKIT_DOTENV")
|
|
if path_hint:
|
|
p = Path(path_hint).resolve()
|
|
if p.exists():
|
|
load_dotenv(dotenv_path=p, override=True)
|
|
os.environ["CRUDKIT_DOTENV_LOADED"] = str(p)
|
|
return
|
|
|
|
repo_env = Path(__file__).resolve().parents[1] / ".env"
|
|
if repo_env.exists():
|
|
load_dotenv(dotenv_path=repo_env, override=True)
|
|
os.environ["CRUDKIT_DOTENV_LOADED"] = str(repo_env)
|
|
return
|
|
|
|
cwd_env = Path.cwd() / ".env"
|
|
if cwd_env.exists():
|
|
load_dotenv(dotenv_path=cwd_env, override=True)
|
|
os.environ["CRUDKIT_DOTENV_LOADED"] = str(cwd_env)
|
|
|
|
def _getenv(name: str, default: Optional[str] = None) -> Optional[str]:
|
|
"""Treat empty strings as missing. Hekos when OS env has DB_BACKEND=''."""
|
|
val = os.getenv(name)
|
|
if val is None or val.strip() == "":
|
|
return default
|
|
return val
|
|
|
|
def build_database_url(
|
|
*,
|
|
backend: Optional[str] = None,
|
|
url: Optional[str] = None,
|
|
user: Optional[str] = None,
|
|
password: Optional[str] = None,
|
|
host: Optional[str] = None,
|
|
port: Optional[str] = None,
|
|
database: Optional[str] = None,
|
|
driver: Optional[str] = None,
|
|
dsn: Optional[str] = None,
|
|
trusted: Optional[bool] = None,
|
|
options: Optional[Dict[str, str]] = None,
|
|
) -> str:
|
|
"""
|
|
Build a SQLAlchemy URL string. If "url" is provided, it wins.
|
|
Supported: sqlite, postgresql, mysql, mssql (pyodbc)
|
|
"""
|
|
|
|
if url:
|
|
return url
|
|
|
|
backend = (backend or "").lower().strip()
|
|
options = options or {}
|
|
|
|
if backend == "sqlite":
|
|
db_path = database or "app.db"
|
|
if db_path == ":memory:":
|
|
return "sqlite:///:memory:"
|
|
return f"sqlite:///{db_path}"
|
|
|
|
if backend in {"postgres", "postgresql"}:
|
|
driver = driver or "psycopg"
|
|
user = user or ""
|
|
password = password or ""
|
|
creds = f"{quote_plus(user)}:{quote_plus(password)}@" if user or password else ""
|
|
host = host or "localhost"
|
|
port = port or "5432"
|
|
database = database or "app"
|
|
qs = ""
|
|
if options:
|
|
qs = "?" + "&".join(f"{k}={quote_plus(v)}" for k, v in options.items())
|
|
return f"postgresql+{driver}://{creds}{host}:{port}/{database}{qs}"
|
|
|
|
if backend == "mysql":
|
|
driver = driver or "pymysql"
|
|
user = user or ""
|
|
password = password or ""
|
|
creds = f"{quote_plus(user)}:{quote_plus(password)}@" if user or password else ""
|
|
host = host or "localhost"
|
|
port = port or "3306"
|
|
database = database or "app"
|
|
qs = ""
|
|
if options:
|
|
qs = "?" + "&".join(f"{k}={quote_plus(v)}" for k, v in options.items())
|
|
return f"mysql+{driver}://{creds}{host}:{port}/{database}{qs}"
|
|
|
|
if backend in {"mssql", "sqlserver", "sqlsrv"}:
|
|
if dsn:
|
|
qs = ""
|
|
if options:
|
|
qs = "?" + "&".join(f"{k}={quote_plus(v)}" for k, v in options.items())
|
|
return f"mssql+pyodbc://@{quote_plus(dsn)}{qs}"
|
|
|
|
driver = driver or "ODBC Driver 18 for SQL Server"
|
|
host = host or "localhost"
|
|
port = port or "1433"
|
|
database = database or "app"
|
|
|
|
if trusted:
|
|
base_opts = {
|
|
"driver": driver,
|
|
"Trusted_Connection": "yes",
|
|
"Encrypt": "yes",
|
|
"TrustServerCertificate": "yes",
|
|
"MARS_Connection": "yes",
|
|
}
|
|
base_opts.update(options)
|
|
qs = "?" + "&".join(f"{k}={quote_plus(v)}" for k, v in base_opts.items())
|
|
return f"mssql+pyodbc://{host}:{port}/{database}{qs}"
|
|
|
|
user = user or ""
|
|
password = password or ""
|
|
creds = f"{quote_plus(user)}:{quote_plus(password)}@" if user or password else ""
|
|
base_opts = {
|
|
"driver": driver,
|
|
"Encrypt": "yes",
|
|
"TrustServerCertificate": "yes",
|
|
"MARS_Connection": "yes",
|
|
}
|
|
base_opts.update(options)
|
|
qs = "?" + "&".join(f"{k}={quote_plus(v)}" for k, v in base_opts.items())
|
|
return f"mssql+pyodbc://{creds}{host}:{port}/{database}{qs}"
|
|
|
|
raise ValueError(f"Unsupported backend: {backend!r}")
|
|
|
|
class Config:
|
|
"""
|
|
CRUDKit config: environment-first with sane defaults.
|
|
Designed to be subclassed by apps, but fine as-is.
|
|
"""
|
|
|
|
_dotenv_loaded = False
|
|
|
|
DEBUG = False
|
|
TESTING = False
|
|
SECRET_KEY = _getenv("SECRET_KEY", "dev-not-secret")
|
|
|
|
if not _dotenv_loaded:
|
|
_load_dotenv_if_present()
|
|
_dotenv_loaded = True
|
|
|
|
DATABASE_URL = build_database_url(
|
|
url=_getenv("DATABASE_URL"),
|
|
backend=_getenv("DB_BACKEND"),
|
|
user=_getenv("DB_USER"),
|
|
password=_getenv("DB_PASS"),
|
|
host=_getenv("DB_HOST"),
|
|
port=_getenv("DB_PORT"),
|
|
database=_getenv("DB_NAME"),
|
|
driver=_getenv("DB_DRIVER"),
|
|
dsn=_getenv("DB_DSN"),
|
|
trusted=bool(int(_getenv("DB_TRUSTED", "0"))),
|
|
options=None,
|
|
)
|
|
|
|
SQLALCHEMY_ECHO = bool(int(os.getenv("DB_ECHO", "0")))
|
|
POOL_SIZE = int(os.getenv("DB_POOL_SIZE", "5"))
|
|
MAX_OVERFLOW = int(os.getenv("DB_MAX_OVERFLOW", "10"))
|
|
POOL_TIMEOUT = int(os.getenv("DB_POOL_TIMEOUT", "30"))
|
|
POOL_RECYCLE = int(os.getenv("DB_POOL_RECYCLE", "1000"))
|
|
POOL_PRE_PING = True
|
|
|
|
SQLITE_PRAGMAS = {
|
|
"journal_mode": os.getenv("SQLITE_JOURNAL_MODE", "WAL"),
|
|
"foreign_keys": os.getenv("SQLITE_FOREIGN_KEYS", "ON"),
|
|
"synchronous": os.getenv("SQLITE_SYNCHRONOUS", "NORMAL"),
|
|
}
|
|
|
|
@classmethod
|
|
def engine_kwargs(cls) -> Dict[str, Any]:
|
|
url = cls.DATABASE_URL
|
|
kwargs: Dict[str, Any] = {
|
|
"echo": cls.SQLALCHEMY_ECHO,
|
|
"pool_pre_ping": cls.POOL_PRE_PING,
|
|
"future": True,
|
|
}
|
|
|
|
if url.startswith("sqlite://"):
|
|
kwargs["connect_args"] = {"check_same_thread": False}
|
|
|
|
kwargs.update(
|
|
{
|
|
"pool_size": cls.POOL_SIZE,
|
|
"max_overflow": cls.MAX_OVERFLOW,
|
|
"pool_timeout": cls.POOL_TIMEOUT,
|
|
"pool_recycle": cls.POOL_RECYCLE,
|
|
}
|
|
)
|
|
return kwargs
|
|
|
|
@classmethod
|
|
def session_kwargs(cls) -> Dict[str, Any]:
|
|
return {
|
|
"autoflush": False,
|
|
"autocommit": False,
|
|
"expire_on_commit": False,
|
|
"future": True,
|
|
}
|
|
|
|
class DevConfig(Config):
|
|
DEBUG = True
|
|
SQLALCHEMY_ECHO = bool(int(os.getenv("DB_ECHO", "1")))
|
|
|
|
class TestConfig(Config):
|
|
TESTING = True
|
|
DATABASE_URL = build_database_url(backend="sqlite", database=":memory:")
|
|
SQLALCHEMY_ECHO = False
|
|
|
|
class ProdConfig(Config):
|
|
DEBUG = False
|
|
SQLALCHEMY_ECHO = bool(int(os.getenv("DB_ECHO", "0")))
|
|
|
|
def get_config(name: str | None) -> Type[Config]:
|
|
"""
|
|
Resolve config by name. None -> environment variable CRUDKIT_ENV or 'dev'.
|
|
"""
|
|
env = (name or os.getenv("CRUDKIT_ENV") or "dev").lower()
|
|
if env in {"prod", "production"}:
|
|
return ProdConfig
|
|
if env in {"test", "testing", "ci"}:
|
|
return TestConfig
|
|
return DevConfig
|