Fix: optimize export

backend/gn_module_export/conf_schema_toml.py

@@ -2,14 +2,13 @@
    Spécification du schéma toml des paramètres de configurations
 """
 
-from marshmallow import fields, Schema
 from geonature.utils.env import ROOT_DIR
+from marshmallow import Schema, fields
 
 export_format_map = {
     "csv": {"mime": "text/csv", "geofeature": False, "label": "CSV"},
     "json": {"mime": "application/json", "geofeature": False, "label": "Json"},
     "geojson": {"mime": "application/json", "geofeature": True, "label": "GeoJson"},
-    "shp": {"mime": "application/zip", "geofeature": True, "label": "ShapeFile"},
     "gpkg": {"mime": "application/zip", "geofeature": True, "label": "GeoPackage"},
 }  # noqa: E133
 

backend/gn_module_export/tests/fixtures.py

@@ -1,9 +1,12 @@
 import pytest
-from geonature.utils.env import db
 from geonature.tests.fixtures import users
-
-from gn_module_export.models import Export, Licences, CorExportsRoles
+from geonature.utils.env import db
 from pypnusershub.db.models import User
+from utils_flask_sqla_geo.generic import GenericQueryGeo
+
+from gn_module_export.models import CorExportsRoles, Export, Licences
+
+EXPORT_SYNTHESE_NAME = "Synthese SINP"
 
 
 @pytest.fixture(scope="function")
@@ -56,3 +59,19 @@ def exports(group_and_user, users):
         "private_group_associated": export_private_group_associated,
         "private_user_associated": export_private_role_associated,
     }
+
+
+@pytest.fixture
+def export_synthese_sinp():
+    return Export.query.filter(Export.label == EXPORT_SYNTHESE_NAME).one()
+
+
+@pytest.fixture
+def export_synthese_sinp_query(export_synthese_sinp):
+    return GenericQueryGeo(
+        db,
+        export_synthese_sinp.view_name,
+        export_synthese_sinp.schema_name,
+        geometry_field=export_synthese_sinp.geometry_field,
+        limit=10,
+    )

backend/gn_module_export/tests/test_utils/test_export.py

@@ -0,0 +1,70 @@
+import csv
+import json
+import tempfile
+
+import fiona
+import pytest
+
+from gn_module_export.utils.export import _export_as_file, export_as_file
+from gn_module_export.tests.fixtures import (
+    export_synthese_sinp,
+    export_synthese_sinp_query,
+)
+
+
+@pytest.mark.usefixtures("temporary_transaction")
+class TestUtilsExport:
+    def test_export_as_file(
+        self, synthese_data, export_synthese_sinp, export_synthese_sinp_query
+    ):
+        with tempfile.NamedTemporaryFile(suffix=".csv") as f:
+            export_as_file(
+                export_synthese_sinp, "csv", f.name, export_synthese_sinp_query
+            )
+
+    def test_export_csv(self, synthese_data, export_synthese_sinp_query):
+        with tempfile.NamedTemporaryFile(suffix=".csv") as f:
+            _export_as_file("csv", f.name, export_synthese_sinp_query)
+            with open(f.name, "r") as csvfile:
+                dialect = csv.Sniffer().sniff(csvfile.read(1024))
+                csvfile.seek(0)
+                reader = csv.DictReader(csvfile, dialect=dialect)
+                ids = {int(row["id_synthese"]) for row in reader}
+        ids_synthese = {
+            synthese["id_synthese"]
+            for synthese in export_synthese_sinp_query.return_query()["items"]
+        }
+        assert ids_synthese == ids
+
+    def test_export_geojson(self, synthese_data, export_synthese_sinp_query):
+        with tempfile.NamedTemporaryFile(suffix=".geojson") as f:
+            _export_as_file("geojson", f.name, export_synthese_sinp_query)
+            with open(f.name, "r") as json_file:
+                res_geojson = json.load(json_file)
+
+        assert res_geojson["type"] == "FeatureCollection"
+        assert len(res_geojson["features"]) > 0
+
+    def test_export_json(self, synthese_data, export_synthese_sinp_query):
+        with tempfile.NamedTemporaryFile(suffix=".json") as f:
+            _export_as_file("json", f.name, export_synthese_sinp_query)
+            with open(f.name, "r") as json_file:
+                res_json = json.load(json_file)
+
+        assert len(res_json) > 0
+
+    def test_export_geopackage(self, synthese_data, export_synthese_sinp_query):
+        file_name = "/tmp/test_fiona.gpkg"
+        _export_as_file("gpkg", file_name, export_synthese_sinp_query)
+
+        with fiona.open(file_name, "r", "GPKG", overwrite=True) as gpkg:
+            result = {data["properties"]["id_synthese"] for data in gpkg}
+
+        # FIXME: obs1, obs2, obs3 are the only synthese data that can be exported
+        # via the view
+        ids_synthese = {
+            value.id_synthese
+            for key, value in synthese_data.items()
+            if key in ["obs1", "obs2", "obs3"]
+        }
+        assert ids_synthese.issubset(result)

backend/gn_module_export/utils/export.py

@@ -0,0 +1,67 @@
+from typing import List, Optional
+
+from flask import current_app
+from geonature.utils.env import db
+from geonature.utils.errors import GeoNatureError
+from ref_geo.utils import get_local_srid
+from utils_flask_sqla_geo.export import (
+    export_csv,
+    export_geojson,
+    export_geopackage,
+    export_json,
+)
+from utils_flask_sqla_geo.generic import GenericQueryGeo
+
+from gn_module_export.models import Export
+
+
+def export_as_file(
+    export: Export, file_format: str, filename: str, generic_query_geo: GenericQueryGeo
+):
+    # TODO Add export.pk_name when available
+    _export_as_file(
+        file_format,
+        filename,
+        generic_query_geo,
+        srid=export.geometry_srid,
+        pk_name=export.view_pk_column,
+    )
+
+
+def _export_as_file(
+    file_format: str,
+    filename: str,
+    generic_query_geo: GenericQueryGeo,
+    pk_name: Optional[str] = None,
+    srid: Optional[int] = None,
+    columns: Optional[List[str]] = [],
+):
+    format_list = [k for k in current_app.config["EXPORTS"]["export_format_map"].keys()]
+
+    if file_format not in format_list:
+        raise GeoNatureError("Unsupported format")
+    if file_format == "gpkg" and srid is None:
+        srid = get_local_srid(db.session)
+
+    schema_class = generic_query_geo.get_marshmallow_schema(pk_name=pk_name)
+
+    if file_format == "gpkg":
+        export_geopackage(
+            query=generic_query_geo.raw_query(),
+            schema_class=schema_class,
+            filename=filename,
+            geometry_field_name=generic_query_geo.geometry_field,
+            srid=srid,
+        )
+        return
+
+    func_dict = {"geojson": export_geojson, "json": export_json, "csv": export_csv}
+
+    with open(filename, "w") as f:
+        func_dict[file_format](
+            query=generic_query_geo.raw_query(),
+            schema_class=schema_class,
+            fp=f,
+            geometry_field_name=generic_query_geo.geometry_field,
+            columns=columns,
+        )