Config feature added.

crudkit/config.py

@@ -0,0 +1,230 @@
+from __future__ import annotations
+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=False)
+            return
+
+    repo_env = Path(__file__).resolve().parents[2] / ".env"
+    if repo_env.exists():
+        load_dotenv(dotenv_path=repo_env, override=False)
+        return
+
+    cwd_env = Path.cwd() / ".env"
+    if cwd_env.exists():
+        load_dotenv(dotenv_path=cwd_env, override=False)
+
+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()
+    optional = 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"postgresel+{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",
+            }
+            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",
+        }
+        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 = os.getenv("SECRET_KEY", "dev-not-secret")
+
+    if not _dotenv_loaded:
+        _load_dotenv_if_present()
+        _dotenv_loaded = True
+
+    DATABASE_URL = build_database_url(
+        url=os.getenv("DATABASE_URL"),
+        backend=os.getenv("DB_BACKEND"),
+        user=os.getenv("DB_USER"),
+        password=os.getenv("DB_PASS"),
+        host=os.getenv("DB_HOST"),
+        port=os.getenv("DB_PORT"),
+        database=os.getenv("DB_NAME"),
+        driver=os.getenv("DB_DRIVER"),
+        dsn=os.getenv("DB_DSN"),
+        trusted=bool(int(os.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_END 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