Using sqlite3 with SpatiaLite Functions

# -- SpatiaLite context: sqlite3 + mod_spatialite 5.x --
import sqlite3
import platform
from typing import Optional

def init_spatialite(db_path: str, ext_name: Optional[str] = None) -> sqlite3.Connection:
    """
    Open a SQLite database, load SpatiaLite, and initialize spatial metadata.
    Returns a connection with all SpatiaLite geometry functions active.

    Args:
        db_path:  Path to the .db or .gpkg file (created if absent).
        ext_name: Override the extension name; auto-detected when None.
    """
    conn = sqlite3.connect(db_path)

    # Required: Python blocks extension loading by default for security.
    conn.enable_load_extension(True)

    if ext_name is None:
        ext_name = "mod_spatialite"
        if platform.system() == "Linux":
            # Some distros install the file as libspatialite.so instead.
            try:
                conn.load_extension(ext_name)
            except sqlite3.OperationalError:
                ext_name = "libspatialite"

    try:
        conn.load_extension(ext_name)
    except sqlite3.OperationalError as exc:
        raise RuntimeError(
            f"Cannot load SpatiaLite extension '{ext_name}'. "
            "Ensure the library is on LD_LIBRARY_PATH (Linux), PATH (Windows), "
            "or DYLD_LIBRARY_PATH (macOS)."
        ) from exc

    # Seed geometry_columns, spatial_ref_sys, and supporting metadata tables.
    # The argument '1' suppresses the EPSG SRS seed data (faster startup).
    conn.execute("SELECT InitSpatialMetaData(1)")

    # WAL mode: prevents write-lock contention during background sync on
    # offline devices. See the connection-pooling guide for multi-reader setup.
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA synchronous=NORMAL")
    # Increase page cache to 16 MB for large geometry buffers in RAM.
    conn.execute("PRAGMA cache_size=-16000")

    return conn

# -- SpatiaLite context: DDL for a spatial table --
def create_features_table(conn: sqlite3.Connection, srid: int = 4326) -> None:
    conn.execute("""
        CREATE TABLE IF NOT EXISTS features (
            id   INTEGER PRIMARY KEY,
            name TEXT NOT NULL
        )
    """)
    # AddGeometryColumn registers the column in geometry_columns and
    # enforces the SRID on every row — required for index creation.
    conn.execute(
        "SELECT AddGeometryColumn('features', 'geom', ?, 'GEOMETRY', 'XY')",
        (srid,)
    )
    conn.commit()

# -- SpatiaLite context: geometry insert via GeomFromText --
def insert_feature(conn: sqlite3.Connection,
                   name: str, wkt: str, srid: int = 4326) -> None:
    conn.execute(
        "INSERT INTO features (name, geom) VALUES (?, GeomFromText(?, ?))",
        (name, wkt, srid)
    )
    conn.commit()

# -- SpatiaLite context: R-tree index creation --
def build_spatial_index(conn: sqlite3.Connection,
                        table: str = "features",
                        col: str = "geom") -> None:
    conn.execute(f"SELECT CreateSpatialIndex('{table}', '{col}')")
    conn.commit()

# -- SpatiaLite context: ST_Buffer + ST_Intersects query --
def find_within_distance(conn: sqlite3.Connection,
                         wkt_point: str,
                         distance_deg: float,
                         srid: int = 4326) -> list[tuple]:
    """
    Return all features whose geometry intersects a circular buffer
    around wkt_point. distance_deg is in decimal degrees for EPSG:4326.
    For metric projections pass a target_srid and use ST_Transform first.
    """
    return conn.execute("""
        SELECT f.id, f.name, ST_AsText(f.geom)
        FROM   features f
        WHERE  ST_Intersects(
                   f.geom,
                   ST_Buffer(GeomFromText(?, ?), ?)
               ) = 1
    """, (wkt_point, srid, distance_deg)).fetchall()

# -- SpatiaLite context: coordinate transformation via ST_Transform --
def reproject_feature(conn: sqlite3.Connection,
                      feature_id: int,
                      target_srid: int) -> tuple | None:
    """
    Returns (id, wkt_in_target_srs) for the given feature.
    Requires PROJ_LIB to point at a valid proj.db.
    """
    return conn.execute("""
        SELECT id, ST_AsText(ST_Transform(geom, ?))
        FROM   features
        WHERE  id = ?
    """, (target_srid, feature_id)).fetchone()

# -- SpatiaLite context: post-init verification --
def verify_spatialite(conn: sqlite3.Connection) -> None:
    # 1. Confirm SpatiaLite version is 5.x
    (version,) = conn.execute("SELECT spatialite_version()").fetchone()
    assert version.startswith("5"), f"Expected SpatiaLite 5.x, got {version}"

    # 2. Confirm geometry_columns table exists (seeded by InitSpatialMetaData)
    (count,) = conn.execute(
        "SELECT COUNT(*) FROM sqlite_master WHERE name='geometry_columns'"
    ).fetchone()
    assert count == 1, "geometry_columns missing — InitSpatialMetaData may have failed"

    # 3. Confirm the spatial index virtual table is present
    (idx_count,) = conn.execute(
        "SELECT COUNT(*) FROM sqlite_master WHERE name='idx_features_geom'"
    ).fetchone()
    assert idx_count == 1, "Spatial index not found — run CreateSpatialIndex first"

    print(f"SpatiaLite {version} OK — metadata and index verified")

# Confirm mod_spatialite loads and returns a version string
sqlite3 /tmp/test.db \
  ".load mod_spatialite" \
  "SELECT spatialite_version();"

# macOS alternative: pip install pysqlite3
import pysqlite3 as sqlite3  # drop-in replacement
# remainder of init_spatialite unchanged

python -c "import os; print(os.environ.get('PROJ_LIB', 'NOT SET'))"