Get the database engine. Args: url: the DB url to use. Returns: The database engine. Raises: ValueError: If the database url is None.
(url: str | None = None)
| 28 | |
| 29 | |
| 30 | def get_engine(url: str | None = None): |
| 31 | """Get the database engine. |
| 32 | |
| 33 | Args: |
| 34 | url: the DB url to use. |
| 35 | |
| 36 | Returns: |
| 37 | The database engine. |
| 38 | |
| 39 | Raises: |
| 40 | ValueError: If the database url is None. |
| 41 | """ |
| 42 | conf = get_config() |
| 43 | url = url or conf.db_url |
| 44 | if url is None: |
| 45 | raise ValueError("No database url configured") |
| 46 | if not Path(constants.ALEMBIC_CONFIG).exists(): |
| 47 | console.warn( |
| 48 | "Database is not initialized, run [bold]nextpy db init[/bold] first." |
| 49 | ) |
| 50 | # Print the SQL queries if the log level is INFO or lower. |
| 51 | echo_db_query = os.environ.get("SQLALCHEMY_ECHO") == "True" |
| 52 | # Needed for the admin dash on sqlite. |
| 53 | connect_args = {"check_same_thread": False} if url.startswith("sqlite") else {} |
| 54 | return sqlmodel.create_engine(url, echo=echo_db_query, connect_args=connect_args) |
| 55 | |
| 56 | |
| 57 | class Model(Base, sqlmodel.SQLModel): |
no test coverage detected