mirror of
https://github.com/simonw/sqlite-utils.git
synced 2026-08-02 15:34:10 +02:00
Add SpatiaLite helpers to CLI (#407)
* Add SpatiaLite CLI helpers * Add docs for spaitalite helpers * Fix flake8 issues and add more detail on spatial types * Run cog and add some help text. * Use SpatiaLite when calculating coverage, refs #407 Co-authored-by: Simon Willison <swillison@gmail.com>
This commit is contained in:
parent
e7f040106b
commit
a692c56659
6 changed files with 370 additions and 35 deletions
|
|
@ -7,7 +7,6 @@ from unittest import mock
|
|||
import json
|
||||
import os
|
||||
import pytest
|
||||
from sqlite_utils.utils import sqlite3, find_spatialite
|
||||
import textwrap
|
||||
|
||||
from .utils import collapse_whitespace
|
||||
|
|
@ -792,34 +791,6 @@ def test_query_raw(db_path, content, is_binary):
|
|||
assert result.output == str(content)
|
||||
|
||||
|
||||
@pytest.mark.skipif(not find_spatialite(), reason="Could not find SpatiaLite extension")
|
||||
@pytest.mark.skipif(
|
||||
not hasattr(sqlite3.Connection, "enable_load_extension"),
|
||||
reason="sqlite3.Connection missing enable_load_extension",
|
||||
)
|
||||
@pytest.mark.parametrize("use_spatialite_shortcut", [True, False])
|
||||
def test_query_load_extension(use_spatialite_shortcut):
|
||||
# Without --load-extension:
|
||||
result = CliRunner().invoke(cli.cli, [":memory:", "select spatialite_version()"])
|
||||
assert result.exit_code == 1
|
||||
assert "no such function: spatialite_version" in result.output
|
||||
# With --load-extension:
|
||||
if use_spatialite_shortcut:
|
||||
load_extension = "spatialite"
|
||||
else:
|
||||
load_extension = find_spatialite()
|
||||
result = CliRunner().invoke(
|
||||
cli.cli,
|
||||
[
|
||||
":memory:",
|
||||
"select spatialite_version()",
|
||||
"--load-extension={}".format(load_extension),
|
||||
],
|
||||
)
|
||||
assert result.exit_code == 0, result.stdout
|
||||
assert ["spatialite_version()"] == list(json.loads(result.output)[0].keys())
|
||||
|
||||
|
||||
def test_query_memory_does_not_create_file(tmpdir):
|
||||
owd = os.getcwd()
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -1,7 +1,10 @@
|
|||
import json
|
||||
import pytest
|
||||
from sqlite_utils.utils import find_spatialite
|
||||
|
||||
from click.testing import CliRunner
|
||||
from sqlite_utils.cli import cli
|
||||
from sqlite_utils.db import Database
|
||||
from sqlite_utils.utils import sqlite3
|
||||
from sqlite_utils.utils import find_spatialite, sqlite3
|
||||
|
||||
pytestmark = [
|
||||
pytest.mark.skipif(
|
||||
|
|
@ -14,6 +17,7 @@ pytestmark = [
|
|||
]
|
||||
|
||||
|
||||
# python API tests
|
||||
def test_find_spatialite():
|
||||
spatialite = find_spatialite()
|
||||
assert spatialite is None or isinstance(spatialite, str)
|
||||
|
|
@ -81,3 +85,152 @@ def test_double_create_spatial_index():
|
|||
|
||||
# call it again, return False
|
||||
assert not table.create_spatial_index("geometry")
|
||||
|
||||
|
||||
# cli tests
|
||||
@pytest.mark.parametrize("use_spatialite_shortcut", [True, False])
|
||||
def test_query_load_extension(use_spatialite_shortcut):
|
||||
# Without --load-extension:
|
||||
result = CliRunner().invoke(cli, [":memory:", "select spatialite_version()"])
|
||||
assert result.exit_code == 1
|
||||
assert "no such function: spatialite_version" in result.output
|
||||
# With --load-extension:
|
||||
if use_spatialite_shortcut:
|
||||
load_extension = "spatialite"
|
||||
else:
|
||||
load_extension = find_spatialite()
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
":memory:",
|
||||
"select spatialite_version()",
|
||||
"--load-extension={}".format(load_extension),
|
||||
],
|
||||
)
|
||||
assert result.exit_code == 0, result.stdout
|
||||
assert ["spatialite_version()"] == list(json.loads(result.output)[0].keys())
|
||||
|
||||
|
||||
def test_cli_create_spatialite(tmpdir):
|
||||
# sqlite-utils create test.db --init-spatialite
|
||||
db_path = tmpdir / "created.db"
|
||||
result = CliRunner().invoke(
|
||||
cli, ["create-database", str(db_path), "--init-spatialite"]
|
||||
)
|
||||
|
||||
assert 0 == result.exit_code
|
||||
assert db_path.exists()
|
||||
assert db_path.read_binary()[:16] == b"SQLite format 3\x00"
|
||||
|
||||
db = Database(str(db_path))
|
||||
assert "spatial_ref_sys" in db.table_names()
|
||||
|
||||
|
||||
def test_cli_add_geometry_column(tmpdir):
|
||||
# create a rowid table with one column
|
||||
db_path = tmpdir / "spatial.db"
|
||||
db = Database(str(db_path))
|
||||
db.init_spatialite()
|
||||
|
||||
table = db["locations"].create({"name": str})
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"add-geometry-column",
|
||||
str(db_path),
|
||||
table.name,
|
||||
"geometry",
|
||||
"--type",
|
||||
"POINT",
|
||||
],
|
||||
)
|
||||
|
||||
assert 0 == result.exit_code
|
||||
|
||||
assert db["geometry_columns"].get(["locations", "geometry"]) == {
|
||||
"f_table_name": "locations",
|
||||
"f_geometry_column": "geometry",
|
||||
"geometry_type": 1, # point
|
||||
"coord_dimension": 2,
|
||||
"srid": 4326,
|
||||
"spatial_index_enabled": 0,
|
||||
}
|
||||
|
||||
|
||||
def test_cli_add_geometry_column_options(tmpdir):
|
||||
# create a rowid table with one column
|
||||
db_path = tmpdir / "spatial.db"
|
||||
db = Database(str(db_path))
|
||||
db.init_spatialite()
|
||||
table = db["locations"].create({"name": str})
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"add-geometry-column",
|
||||
str(db_path),
|
||||
table.name,
|
||||
"geometry",
|
||||
"-t",
|
||||
"POLYGON",
|
||||
"--srid",
|
||||
"3857", # https://epsg.io/3857
|
||||
"--not-null",
|
||||
],
|
||||
)
|
||||
|
||||
assert 0 == result.exit_code
|
||||
|
||||
assert db["geometry_columns"].get(["locations", "geometry"]) == {
|
||||
"f_table_name": "locations",
|
||||
"f_geometry_column": "geometry",
|
||||
"geometry_type": 3, # polygon
|
||||
"coord_dimension": 2,
|
||||
"srid": 3857,
|
||||
"spatial_index_enabled": 0,
|
||||
}
|
||||
|
||||
column = table.columns[1]
|
||||
assert column.notnull
|
||||
|
||||
|
||||
def test_cli_add_geometry_column_invalid_type(tmpdir):
|
||||
# create a rowid table with one column
|
||||
db_path = tmpdir / "spatial.db"
|
||||
db = Database(str(db_path))
|
||||
db.init_spatialite()
|
||||
|
||||
table = db["locations"].create({"name": str})
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli,
|
||||
[
|
||||
"add-geometry-column",
|
||||
str(db_path),
|
||||
table.name,
|
||||
"geometry",
|
||||
"--type",
|
||||
"NOT-A-TYPE",
|
||||
],
|
||||
)
|
||||
|
||||
assert 2 == result.exit_code
|
||||
|
||||
|
||||
def test_cli_create_spatial_index(tmpdir):
|
||||
# create a rowid table with one column
|
||||
db_path = tmpdir / "spatial.db"
|
||||
db = Database(str(db_path))
|
||||
db.init_spatialite()
|
||||
|
||||
table = db["locations"].create({"name": str})
|
||||
table.add_geometry_column("geometry", "POINT")
|
||||
|
||||
result = CliRunner().invoke(
|
||||
cli, ["create-spatial-index", str(db_path), table.name, "geometry"]
|
||||
)
|
||||
|
||||
assert 0 == result.exit_code
|
||||
|
||||
assert "idx_locations_geometry" in db.table_names()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue