Get the database engine arguments. Args: url: The database url. Returns: The database engine arguments as a dict.
(url: str | None = None)
| 61 | _ASYNC_ENGINE: dict[str, sqlalchemy.ext.asyncio.AsyncEngine] = {} |
| 62 | |
| 63 | def get_engine_args(url: str | None = None) -> dict[str, Any]: |
| 64 | """Get the database engine arguments. |
| 65 | |
| 66 | Args: |
| 67 | url: The database url. |
| 68 | |
| 69 | Returns: |
| 70 | The database engine arguments as a dict. |
| 71 | """ |
| 72 | kwargs: dict[str, Any] = { |
| 73 | # Print the SQL queries if the log level is INFO or lower. |
| 74 | "echo": environment.SQLALCHEMY_ECHO.get(), |
| 75 | # Check connections before returning them. |
| 76 | "pool_pre_ping": environment.SQLALCHEMY_POOL_PRE_PING.get(), |
| 77 | "pool_size": environment.SQLALCHEMY_POOL_SIZE.get(), |
| 78 | "max_overflow": environment.SQLALCHEMY_MAX_OVERFLOW.get(), |
| 79 | "pool_recycle": environment.SQLALCHEMY_POOL_RECYCLE.get(), |
| 80 | "pool_timeout": environment.SQLALCHEMY_POOL_TIMEOUT.get(), |
| 81 | } |
| 82 | conf = get_config() |
| 83 | url = url or conf.db_url |
| 84 | if url is not None and url.startswith("sqlite"): |
| 85 | # Needed for the admin dash on sqlite. |
| 86 | kwargs["connect_args"] = {"check_same_thread": False} |
| 87 | return kwargs |
| 88 | |
| 89 | def get_engine(url: str | None = None) -> sqlalchemy.engine.Engine: |
| 90 | """Get the database engine. |
no test coverage detected