feat: added in env.py the way to get setted sqlalchemy.url

alembic_migration/README.md

@@ -32,7 +32,7 @@ To use the databases, set the following environment variables:
 
 ```sh
 TDP_ALEMBIC_SQLITE_DSN=sqlite:///sqlite.db
-TDP_ALEMBIC_POSGRESQL_DSN=postgresql://postgres:postgres@localhost:5432/tdp
+TDP_ALEMBIC_POSTGRESQL_DSN=postgresql://postgres:postgres@localhost:5432/tdp
 TDP_ALEMBIC_MYSQL_DSN=mysql+pymysql://mysql:mysql@localhost:3306/tdp
 ```
 

alembic_migration/mysql/env.py

@@ -2,11 +2,9 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import os
-from pathlib import Path
 
 from alembic import context
-from dotenv import load_dotenv
-from sqlalchemy import create_engine
+from sqlalchemy import engine_from_config, pool
 
 from tdp.core.models.base_model import BaseModel
 
@@ -18,6 +16,18 @@
 target_metadata = BaseModel.metadata
 
 
+def load_dotenv():
+    """Load the .env file based on the TDP_ENV environment variable."""
+    import os
+    from pathlib import Path
+
+    from dotenv import load_dotenv
+
+    env_path = Path(os.environ.get("TDP_ENV", ".env"))
+    if env_path.exists:
+        load_dotenv(env_path)
+
+
 def run_migrations_offline() -> None:
     """Run migrations in 'offline' mode. Not implemented."""
     raise NotImplementedError()
@@ -26,17 +36,23 @@ def run_migrations_offline() -> None:
 def run_migrations_online() -> None:
     """Run migrations in 'online' mode."""
 
-    env_path = Path(os.environ.get("TDP_ENV", ".env"))
-    if env_path.exists:
-        load_dotenv(env_path)
-
-    database_dsn = os.environ.get("TDP_ALEMBIC_MYSQL_DSN")
+    # Get the configuration based on the current section name
+    engine_config = config.get_section(config.config_ini_section, {})
 
-    if database_dsn is None:
-        raise ValueError("TDP_ALEMBIC_MYSQL_DSN env var is missing")
+    # Override the database URL with the one from the environment if not set
+    if not "sqlalchemy.url" in engine_config:
+        load_dotenv()
+        if not (database_dsn := os.environ.get("TDP_ALEMBIC_MYSQL_DSN")):
+            raise ValueError("TDP_ALEMBIC_MYSQL_DSN env var is missing")
+        engine_config["sqlalchemy.url"] = database_dsn
 
-    connectable = create_engine(database_dsn)
+    connectable = engine_from_config(
+        engine_config,
+        prefix="sqlalchemy.",
+        poolclass=pool.NullPool,
+    )
 
+    # Ensure that the provided database url is of the proper dialect
     if connectable.dialect.name != "mysql":
         raise ValueError("You are not connected to a MySQL or MariaDB database")
 

alembic_migration/postgresql/env.py

@@ -2,12 +2,10 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import os
-from pathlib import Path
 
 import alembic_postgresql_enum  # noqa: F401
 from alembic import context
-from dotenv import load_dotenv
-from sqlalchemy import create_engine
+from sqlalchemy import engine_from_config, pool
 
 from tdp.core.models.base_model import BaseModel
 
@@ -19,24 +17,43 @@
 target_metadata = BaseModel.metadata
 
 
+def load_dotenv():
+    """Load the .env file based on the TDP_ENV environment variable."""
+    import os
+    from pathlib import Path
+
+    from dotenv import load_dotenv
+
+    env_path = Path(os.environ.get("TDP_ENV", ".env"))
+    if env_path.exists:
+        load_dotenv(env_path)
+
+
 def run_migrations_offline() -> None:
     """Run migrations in 'offline' mode. Not implemented."""
     raise NotImplementedError()
 
 
 def run_migrations_online() -> None:
     """Run migrations in 'online' mode."""
-    env_path = Path(os.environ.get("TDP_ENV", ".env"))
-    if env_path.exists:
-        load_dotenv(env_path)
 
-    database_dsn = os.environ.get("TDP_ALEMBIC_POSGRESQL_DSN")
+    # Get the configuration based on the current section name
+    engine_config = config.get_section(config.config_ini_section, {})
 
-    if database_dsn is None:
-        raise ValueError("TDP_ALEMBIC_POSGRESQL_DSN env var is missing")
+    # Override the database URL with the one from the environment if not set
+    if not "sqlalchemy.url" in engine_config:
+        load_dotenv()
+        if not (database_dsn := os.environ.get("TDP_ALEMBIC_POSTGRESQL_DSN")):
+            raise ValueError("TDP_ALEMBIC_POSTGRESQL_DSN env var is missing")
+        engine_config["sqlalchemy.url"] = database_dsn
 
-    connectable = create_engine(database_dsn)
+    connectable = engine_from_config(
+        engine_config,
+        prefix="sqlalchemy.",
+        poolclass=pool.NullPool,
+    )
 
+    # Ensure that the provided database url is of the proper dialect
     if connectable.dialect.name != "postgresql":
         raise ValueError("You are not connected to a PostgreSQL database")
 

alembic_migration/sqlite/env.py

@@ -2,11 +2,9 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import os
-from pathlib import Path
 
 from alembic import context
-from dotenv import load_dotenv
-from sqlalchemy import create_engine
+from sqlalchemy import engine_from_config, pool
 
 from tdp.core.models.base_model import BaseModel
 
@@ -18,6 +16,18 @@
 target_metadata = BaseModel.metadata
 
 
+def load_dotenv():
+    """Load the .env file based on the TDP_ENV environment variable."""
+    import os
+    from pathlib import Path
+
+    from dotenv import load_dotenv
+
+    env_path = Path(os.environ.get("TDP_ENV", ".env"))
+    if env_path.exists:
+        load_dotenv(env_path)
+
+
 def run_migrations_offline() -> None:
     """Run migrations in 'offline' mode. Not implemented."""
     raise NotImplementedError()
@@ -26,17 +36,23 @@ def run_migrations_offline() -> None:
 def run_migrations_online() -> None:
     """Run migrations in 'online' mode."""
 
-    env_path = Path(os.environ.get("TDP_ENV", ".env"))
-    if env_path.exists:
-        load_dotenv(env_path)
-
-    database_dsn = os.environ.get("TDP_ALEMBIC_SQLITE_DSN")
+    # Get the configuration based on the current section name
+    engine_config = config.get_section(config.config_ini_section, {})
 
-    if database_dsn is None:
-        raise ValueError("TDP_ALEMBIC_SQLITE_DSN env var is missing")
+    # Override the database URL with the one from the environment if not set
+    if not "sqlalchemy.url" in engine_config:
+        load_dotenv()
+        if not (database_dsn := os.environ.get("TDP_ALEMBIC_SQLITE_DSN")):
+            raise ValueError("TDP_ALEMBIC_SQLITE_DSN env var is missing")
+        engine_config["sqlalchemy.url"] = database_dsn
 
-    connectable = create_engine(database_dsn)
+    connectable = engine_from_config(
+        engine_config,
+        prefix="sqlalchemy.",
+        poolclass=pool.NullPool,
+    )
 
+    # Ensure that the provided database url is of the proper dialect
     if connectable.dialect.name != "sqlite":
         raise ValueError("You are not connected to an SQLite database")