Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions CHANGES.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,11 @@
# CHANGELOG

## 0.13.0 (xxxx-xx-xx)

### Improvements

- Add whether append is supported in `list_drivers` (#559)

## 0.12.0 (xxxx-xx-xx)

### Potentially breaking changes
Expand Down
1 change: 1 addition & 0 deletions pyogrio/_compat.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
GDAL_GE_37 = __gdal_version__ >= (3, 7, 0)
GDAL_GE_38 = __gdal_version__ >= (3, 8, 0)
GDAL_GE_311 = __gdal_version__ >= (3, 11, 0)
GDAL_GE_312 = __gdal_version__ >= (3, 12, 0)

HAS_GDAL_GEOS = __gdal_geos_version__ is not None

Expand Down
32 changes: 28 additions & 4 deletions pyogrio/_ogr.pyx
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,24 @@ def get_gdal_config_option(str name):
return str_value


def ogr_driver_supports_update(driver):
# check metadata for driver to see if it supports update
IF CTE_GDAL_VERSION >= (3, 11, 0):
if _get_driver_metadata_item(driver, "DCAP_UPDATE") == "YES":
return True

return False


def ogr_driver_supports_append(driver):
# check metadata for driver to see if it supports append
IF CTE_GDAL_VERSION >= (3, 12, 0):
if _get_driver_metadata_item(driver, "DCAP_APPEND") == "YES":
return True

return False


def ogr_driver_supports_write(driver):
# check metadata for driver to see if it supports write
if _get_driver_metadata_item(driver, "DCAP_CREATE") == "YES":
Expand All @@ -121,14 +139,20 @@ def ogr_list_drivers():
for i in range(OGRGetDriverCount()):
driver = OGRGetDriver(i)
name_c = <char *>OGR_Dr_GetName(driver)

name = get_string(name_c)

if ogr_driver_supports_write(name):
drivers[name] = "rw"
capability = "r"

if ogr_driver_supports_update(name):
capability += "a"
else:
drivers[name] = "r"
if ogr_driver_supports_append(name):
capability += "a"

if ogr_driver_supports_write(name):
capability += "w"

drivers[name] = capability

return drivers

Expand Down
14 changes: 12 additions & 2 deletions pyogrio/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@
__gdal_geos_version__ = get_gdal_geos_version()


def list_drivers(read=False, write=False):
def list_drivers(read=False, write=False, append=False):
"""List drivers available in GDAL.

Parameters
Expand All @@ -50,13 +50,20 @@ def list_drivers(read=False, write=False):
If True, will only return drivers that are known to support read capabilities.
write: bool, optional (default: False)
If True, will only return drivers that are known to support write capabilities.
append: bool, optional (default: False)
If True, will only return drivers that are known to support append capabilities.
.. versionadded:: 0.13.0

Returns
-------
dict
Mapping of driver name to file mode capabilities: ``"r"``: read, ``"w"``: write.
Mapping of driver name to file mode capabilities: ``"r"``: read,
``"a"``: append, ``"w"``: write.
Drivers that are available but with unknown support are marked with ``"?"``

.. versionchanged:: 0.13.0
Added the ``a`` flag, which is available for GDAL >= 3.11.

"""
drivers = ogr_list_drivers()

Expand All @@ -66,6 +73,9 @@ def list_drivers(read=False, write=False):
if write:
drivers = {k: v for k, v in drivers.items() if v.endswith("w")}

if append:
drivers = {k: v for k, v in drivers.items() if "a" in v}

return drivers


Expand Down
65 changes: 60 additions & 5 deletions pyogrio/tests/test_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
vsi_rmtree,
vsi_unlink,
)
from pyogrio._compat import GDAL_GE_38
from pyogrio._compat import GDAL_GE_38, GDAL_GE_312
from pyogrio._env import GDALEnv
from pyogrio.errors import DataLayerError, DataSourceError
from pyogrio.raw import read, write
Expand Down Expand Up @@ -137,16 +137,63 @@ def test_ogr_driver_supports_write(driver, expected):
def test_list_drivers():
all_drivers = list_drivers()

# Expected capabilities based on `fiona.supported_drivers`.
expected_drivers: dict[str, str] = {
"AeronavFAA": "r",
"ARCGEN": "r",
"BNA": "rw",
"DXF": "rw",
"CSV": "raw",
"FileGDB": "raw",
"OpenFileGDB": "raw",
"ESRIJSON": "r",
"ESRI Shapefile": "raw",
"FlatGeobuf": "rw", # Changed: "raw" to "rw": append only if no spatial index
"GeoJSON": "raw",
"GeoJSONSeq": "raw",
"GPKG": "raw",
"GML": "rw",
"GMT": "rw",
"OGR_GMT": "rw",
"GPX": "rw",
"Idrisi": "r",
"MapInfo File": "raw",
"DGN": "rw", # Changed: "raw" to "rw": unclear if append is possible
"Parquet": "rw",
"PCIDSK": "raw",
"PDS": "r",
"OGR_PDS": "r",
"S57": "rw", # Changed: "r" to "rw": create supported according to GDAL docs
"SEGY": "r",
"SQLite": "raw",
"SUA": "r",
"TileDB": "raw",
"TopoJSON": "r",
}

# verify that the core drivers are present
for name in ("ESRI Shapefile", "GeoJSON", "GeoJSONSeq", "GPKG", "OpenFileGDB"):
assert name in all_drivers
expected_capability = "rw"
assert all_drivers[name] == expected_capability
for name, expected_capability in expected_drivers.items():
if name not in all_drivers:
print(f"{name} not in list_drivers(), ignore")
continue

if not GDAL_GE_312:
expected_capability = expected_capability.replace("a", "")
if name == "OpenFileGDB" and __gdal_version__ < (3, 6, 0):
expected_capability = "r"

assert all_drivers[name] == expected_capability, (
f"Error for {name}: {expected_capability=}, {all_drivers[name]=}"
)

drivers = list_drivers(read=True)
expected = {k: v for k, v in all_drivers.items() if v.startswith("r")}
assert len(drivers) == len(expected)

drivers = list_drivers(append=True)
expected = {k: v for k, v in all_drivers.items() if "a" in v}
assert len(drivers) == len(expected)

drivers = list_drivers(write=True)
expected = {k: v for k, v in all_drivers.items() if v.endswith("w")}
assert len(drivers) == len(expected)
Expand All @@ -157,6 +204,14 @@ def test_list_drivers():
}
assert len(drivers) == len(expected)

drivers = list_drivers(read=True, write=True, append=True)
expected = {
k: v
for k, v in all_drivers.items()
if v.startswith("r") and v.endswith("w") and "a" in v
}
assert len(drivers) == len(expected)


def test_list_layers(
naturalearth_lowres,
Expand Down
Loading