Merge branch 'main' into mapproxy

Author: nicokant

.pre-commit-config.yaml

@@ -67,3 +67,12 @@ repos:
         language: system
         types: [python]
         pass_filenames: false
+
+  - repo: local
+    hooks:
+      - id: ty
+        name: ty check
+        entry: uv run ty check src
+        language: system
+        types: [python]
+        pass_filenames: false

.vscode/extensions.json

@@ -1,6 +1,7 @@
 {
     "recommendations": [
         "charliermarsh.ruff",
-        "elagil.pre-commit-helper"
+        "elagil.pre-commit-helper",
+        "astral-sh.ty"
     ]
 }

pyproject.toml

@@ -5,9 +5,11 @@ requires = ["hatchling"]
 [dependency-groups]
 dev = [
   "deptry",
+  "lxml-stubs>=0.5.1",
   "mkdocs>=1.6.1",
   "mkdocs-material>=9.7.1",
-  "mkdocs-typer2>=0.1.6"
+  "mkdocs-typer2>=0.1.6",
+  "ty>=0.0.3"
 ]
 
 [project]

src/datasync/dms.py

@@ -28,7 +28,7 @@ def guess_type(driver: str) -> str:
     return "image/tiff"
 
 
-def to_iso19139(metadata: dict) -> str:
+def to_iso19139(metadata: str) -> str | dict:
     loaded = orjson.loads(metadata)
     log.debug(loaded, id=loaded.get("identification", {}).get("identifier"))
     try:
@@ -123,7 +123,10 @@ def generate_csw_metadata(
 
     descriptions_structure = datasets.aggregate(
         "json_group_structure(metadata->'$.descriptions[*]') as stucture"
-    ).fetchone()[0]
+    ).fetchone()
+    descriptions_structure = (
+        descriptions_structure[0] if descriptions_structure else None
+    )
 
     log.debug("Generated schema of abstract", structure=descriptions_structure)
 
@@ -438,7 +441,10 @@ def generate_geoapi_config(
 
     descriptions_structure = datasets.aggregate(
         "json_group_structure(metadata->'$.descriptions[*]') as stucture"
-    ).fetchone()[0]
+    ).fetchone()
+    descriptions_structure = (
+        descriptions_structure[0] if descriptions_structure else None
+    )
 
     descriptions = (
         (

src/datasync/gbif_backbone.py

@@ -0,0 +1,65 @@
+import fsspec
+import typer
+from duckdb import (
+    DuckDBPyConnection,
+    connect,
+)
+
+from .libs.helpers import DuckDBAtomicTransaction
+from .settings import (
+    env,
+    log,
+)
+
+log.debug("Importing GBIF Backbone settings")
+
+
+GBIF_BACKBONE_DUCKDB_NAME = env.path(
+    "GBIF_BACKBONE_DUCKDB_FILE_NAME", default="gbif_backbone.duckdb"
+).root
+GBIF_BACKBONE_URL = env.url(
+    "GBIF_BACKBONE_URL",
+    default="https://hosted-datasets.gbif.org/datasets/backbone/current/backbone.zip",
+)
+
+app = typer.Typer(help="export GBIF Backbone data to DuckDB database")
+
+
+def import_taxon(conn: DuckDBPyConnection, archive):
+    log.debug("Importing Taxon.tsv")
+    conn.sql("DROP TABLE IF EXISTS taxon")
+    conn.from_csv_auto(archive.open("Taxon.tsv")).to_table("taxon")
+    conn.sql("CREATE INDEX taxon_taxonid ON taxon (taxonid)")
+    conn.sql("""
+        PRAGMA create_fts_index(
+            "taxon", "taxonID", "canonicalName", overwrite=TRUE
+        )
+    """)
+
+
+def import_vernacular_name(conn: DuckDBPyConnection, archive):
+    log.debug("Importing VernacularName.tsv")
+    conn.sql("DROP TABLE IF EXISTS vernacular_name")
+    vernacular_names = conn.from_csv_auto(archive.open("VernacularName.tsv"))  # noqa: F841
+    conn.sql("""
+        SELECT *, CONCAT_WS('|', taxonID, language, vernacularName) AS vernacularID
+        FROM vernacular_names
+    """).to_table("vernacular_name")
+    conn.sql("CREATE INDEX vernacular_name_taxonid ON vernacular_name (taxonid)")
+    conn.sql("CREATE INDEX vernacular_name_language ON vernacular_name (language)")
+    conn.sql("""
+        PRAGMA create_fts_index(
+            "vernacular_name", "vernacularID", "vernacularName", overwrite=TRUE
+        )
+    """)
+
+
+@app.command()
+def import_all():
+    """Import GBIF Backbone data into a DuckDB database."""
+    archive = fsspec.filesystem("zip", fo=GBIF_BACKBONE_URL.geturl(), mode="r")
+    with connect(GBIF_BACKBONE_DUCKDB_NAME) as conn:
+        with DuckDBAtomicTransaction(conn):
+            import_taxon(conn, archive)
+            import_vernacular_name(conn, archive)
+    log.info("GBIF Backbone data imported successfully")

src/datasync/grass.py

@@ -59,25 +59,25 @@ def register_layers(parquet_file_path: str, project_number: str, gisbase: str):
                     "split_part",
                     duckdb.ColumnExpression("cleaned_file"),
                     duckdb.ConstantExpression("/"),
-                    duckdb.ConstantExpression(1),
+                    duckdb.ConstantExpression("1"),
                 ).alias("location"),
                 duckdb.FunctionExpression(
                     "split_part",
                     duckdb.ColumnExpression("cleaned_file"),
                     duckdb.ConstantExpression("/"),
-                    duckdb.ConstantExpression(2),
+                    duckdb.ConstantExpression("2"),
                 ).alias("mapset"),
                 duckdb.FunctionExpression(
                     "split_part",
                     duckdb.ColumnExpression("cleaned_file"),
                     duckdb.ConstantExpression("/"),
-                    duckdb.ConstantExpression(3),
+                    duckdb.ConstantExpression("3"),
                 ).alias("type"),
                 duckdb.FunctionExpression(
                     "split_part",
                     duckdb.ColumnExpression("cleaned_file"),
                     duckdb.ConstantExpression("/"),
-                    duckdb.ConstantExpression(4),
+                    duckdb.ConstantExpression("4"),
                 ).alias("resource"),
             ]
         )

src/datasync/ipt/csw.py

@@ -6,7 +6,8 @@
 from pygeometa.schemas.iso19139 import ISO19139OutputSchema
 from shapely.geometry import box
 
-from ..settings import env as logger
+from ..libs.helpers import get_anytext
+from ..settings import log
 from .settings import (
     AWS_ENDPOINT_URL,
     CSW_PATH,
@@ -22,29 +23,16 @@
 iso = ISO19139OutputSchema()
 
 
-def get_anytext(bag):
-    """
-    generate bag of text for free text searches
-    accepts list of words, string of XML, or etree.Element
-    """
-
-    if isinstance(bag, list):  # list of words
-        return " ".join([_f for _f in bag if _f]).strip()
-    else:  # xml
-        if isinstance(bag, bytes) or isinstance(bag, str):
-            # serialize to lxml
-            bag = etree.fromstring(bag, PARSER)  # noqa: S320
-        # get all XML element content
-        return " ".join([value.strip() for value in bag.xpath("//text()")])
-
-
 def eml_to_record(ds, text):
-    metadata = eml.import_(text)
+    metadata: dict = eml.import_(text)
 
     metadata["metadata"]["identifier"] = f"ipt__{ds['id']}"
 
     xml = iso.write(metadata)
-    fts = get_anytext(xml)
+    if isinstance(xml, str):
+        fts = get_anytext(xml)
+    else:
+        raise TypeError("xml instance expected to be of type string")
     idf = metadata["identification"]
     bbox = idf["extents"]["spatial"][0]["bbox"]
 
@@ -111,9 +99,9 @@ def eml_to_record(ds, text):
 
 
 def write_eml_record(rows):
-    logger.info("converting to arrow")
+    log.info("converting to arrow")
     records = pa.Table.from_pylist(rows)  # noqa: F841
-    logger.info("write to S3")
+    log.info("write to S3")
     conn.sql("from records").write_parquet(
         f"s3://{S3_BUCKET}{CSW_PATH}",
         compression="zstd",

src/datasync/libs/helpers.py

@@ -1,25 +1,47 @@
 import re
 
+from duckdb import DuckDBPyConnection
 from lxml import etree
+from lxml.etree import _Element
 
-PARSER = etree.XMLParser(resolve_entities=False)
+PARSER: etree.XMLParser = etree.XMLParser(resolve_entities=False)
 
 
-def get_anytext(bag: str) -> str:
+def get_anytext(bag: str | _Element | list[str]) -> str:
     """
     generate bag of text for free text searches
     accepts list of words, string of XML, or etree.Element
     """
 
     if isinstance(bag, list):  # list of words
-        return " ".join([_f for _f in bag if _f]).strip()
+        return " ".join([_f for _f in bag if _f and isinstance(_f, str)]).strip()
     else:  # xml
         if isinstance(bag, bytes) or isinstance(bag, str):
             # serialize to lxml
             bag = etree.fromstring(bag, PARSER)  # noqa: S320
         # get all XML element content
-        return re.sub(
-            r"\s+",
-            " ",
-            " ".join([value.strip() for value in bag.xpath("//text()")]).strip(),
-        )
+        all_text = bag.xpath("//text()")
+        if isinstance(all_text, list):
+            return re.sub(
+                r"\s+",
+                " ",
+                " ".join([str(value).strip() for value in all_text]).strip(),
+            )
+        # NOTE: this should never happen as the xpath evaluation always returns a list
+        # but the type annotation is generic as xpath might return any type
+        raise TypeError("xpath result was not a list of strings")
+
+
+class DuckDBAtomicTransaction:
+    def __init__(self, conn: DuckDBPyConnection):
+        self.conn = conn
+
+    def __enter__(self):
+        self.conn.begin()
+        return self.conn
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        if exc_type is None:
+            self.conn.commit()
+        else:
+            self.conn.rollback()

src/datasync/main.py

@@ -4,7 +4,17 @@
 
 import typer
 
-from . import dms, grass, maps, ninagen, nva, pit_registering_salmon, services, ubw
+from . import (
+    dms,
+    gbif_backbone,
+    grass,
+    maps,
+    ninagen,
+    nva,
+    pit_registering_salmon,
+    services,
+    ubw,
+)
 
 app = typer.Typer(
     help="Provide subcommands for synchronizing different resources, see subcommands"
@@ -17,6 +27,7 @@
 app.add_typer(grass.app, name="grass-gis")
 app.add_typer(services.app, name="services")
 app.add_typer(maps.app, name="maps")
+app.add_typer(gbif_backbone.app, name="gbif-backbone")
 
 if __name__ == "__main__":
     app()

src/datasync/ninagen/snp_analysis.py

@@ -25,7 +25,7 @@ def snp_analysis_to_parquet(file: str) -> None:
                 "\n".join(
                     f.read().replace("\r", "").split("\n\n")[2].split("\n")[2:]
                 ).encode()
-            )
+            ).read()
         )
 
     table.select("* rename (column00 as position, column01 as genlab_id)").query(