Skip to content
Louis-Gm/phytospatial

Setup Db

initialize_database(db_type, path=None, reset=False, env_path=None)

Orchestrates the provisioning of the spatial database schema based on the dialect preference.

Parameters:

Name Type Description Default
db_type str

The requested database dialect ('sqlite' or 'postgres').

 required
path Optional[str]

Optional file path for local SQLite deployments. Defaults to 'phytospatial_local.gpkg'.

 None
reset bool

Instructs the engine to delete an existing local database file or drop all tables.

 False
env_path Optional[Union[str, Path]]

Filepath to a .env configuration file containing database credentials.

 None

Raises:

Type Description
SystemExit

If database dependencies are missing or the schema deployment fails.
Source code in src/phytospatial/db/setup_db.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def initialize_database(
    db_type: str, 
    path: Optional[str] = None, 
    reset: bool = False,
    env_path: Optional[Union[str, Path]] = None
) -> None:
    """
    Orchestrates the provisioning of the spatial database schema based on the dialect preference.

    Args:
        db_type (str): The requested database dialect ('sqlite' or 'postgres').
        path (Optional[str]): Optional file path for local SQLite deployments. Defaults to 'phytospatial_local.gpkg'.
        reset (bool): Instructs the engine to delete an existing local database file or drop all tables.
        env_path (Optional[Union[str, Path]]): Filepath to a .env configuration file containing database credentials.

    Raises:
        SystemExit: If database dependencies are missing or the schema deployment fails.
    """
    try:
        from phytospatial.db.client import DB_Client
        from phytospatial.db.models import Base
    except ImportError as e:
        log.error(f"Database dependencies are missing. Run `pip install phytospatial[db]`. Details: {e}")
        sys.exit(1)

    sqlite_path = path or "phytospatial_local.gpkg"

    if reset and db_type == "sqlite" and Path(sqlite_path).exists():
        log.warning(f"Reset flag detected. Removing existing GeoPackage at {sqlite_path}.")
        os.remove(sqlite_path)

    log.info(f"Targeting database connection: {db_type.upper()}")

    try:
        client = DB_Client.from_env(db_type=db_type, env_path=env_path, sqlite_path=sqlite_path)

        if reset and db_type == "postgres":
            log.warning("Reset flag detected for PostgreSQL. Dropping all existing tables with CASCADE.")
            with client.engine.begin() as conn:
                for table in reversed(Base.metadata.sorted_tables):
                    conn.execute(text(f"DROP TABLE IF EXISTS {table.name} CASCADE"))

        success = client.deploy_schema()

        if success:
            log.info("Database initialization sequence completed successfully.")
        else:
            log.error("Database initialization encountered an error.")
            sys.exit(1)

    except Exception as e:
        log.error(f"Failed to instantiate DB_Client or deploy schema: {e}")
        sys.exit(1)