(database_url: str)
| 52 | |
| 53 | |
| 54 | def get_database_pool_settings(database_url: str) -> Dict[str, Any]: |
| 55 | backend = get_database_backend(database_url) |
| 56 | settings: Dict[str, Any] = { |
| 57 | "backend": backend, |
| 58 | "pool_pre_ping": _get_env_bool("APP_DB_POOL_PRE_PING", True), |
| 59 | } |
| 60 | if backend == "sqlite": |
| 61 | settings["sqlite_check_same_thread"] = False |
| 62 | return settings |
| 63 | |
| 64 | settings.update({ |
| 65 | "pool_size": _get_env_int("APP_DB_POOL_SIZE", 20, 1), |
| 66 | "max_overflow": _get_env_int("APP_DB_MAX_OVERFLOW", 20, 0), |
| 67 | "pool_timeout": _get_env_int("APP_DB_POOL_TIMEOUT", 30, 1), |
| 68 | "pool_recycle": _get_env_int("APP_DB_POOL_RECYCLE", 1800, 30), |
| 69 | "pool_use_lifo": _get_env_bool("APP_DB_POOL_USE_LIFO", True), |
| 70 | }) |
| 71 | return settings |
| 72 | |
| 73 | |
| 74 | def _build_engine_options(database_url: str) -> Dict[str, Any]: |
no test coverage detected