Feat export layer

django_project/cloud_native_gis/models/layer.py

@@ -4,6 +4,8 @@
 import os
 import subprocess
 import uuid
+import zipfile
+from pathlib import Path
 
 from django.conf import settings
 from django.contrib.auth import get_user_model
@@ -18,7 +20,8 @@
 )
 from cloud_native_gis.models.style import Style
 from cloud_native_gis.utils.connection import delete_table
-from cloud_native_gis.utils.fiona import FileType, list_layers
+from cloud_native_gis.utils.type import FileType
+from cloud_native_gis.utils.fiona import list_layers
 
 FOLDER_FILES = 'cloud_native_gis_files'
 PMTILES_FOLDER = 'pmtile_files'
@@ -230,7 +233,6 @@ def _convert_to_geojson(self, layer_upload, layer_files: list):
         subprocess.run(cmd, check=True)
         return file_type, json_filepath
 
-
     def generate_pmtiles(self):
         """
         Generate PMTiles for the current layer.
@@ -335,6 +337,104 @@ def generate_pmtiles(self):
                     f"Failed to generate PMTiles for layer '{self.name}'."
                 )
 
+    def _zip_shapefile(self, shp_filepath, working_dir, remove_file=True):
+        zip_filepath = os.path.join(
+            working_dir,
+            shp_filepath.replace('.shp', '.zip')
+        )
+        file_name = os.path.basename(shp_filepath).replace('.shp', '')
+        shp_files = ['.shp', '.dbf', '.shx', '.cpg', '.prj']
+        with zipfile.ZipFile(
+                zip_filepath, 'w', zipfile.ZIP_DEFLATED) as archive:
+            for suffix in shp_files:
+                shape_file = os.path.join(
+                    working_dir,
+                    file_name
+                ) + suffix
+                if not os.path.exists(shape_file):
+                    continue
+                archive.write(
+                    shape_file,
+                    arcname=file_name + suffix
+                )
+                if remove_file:
+                    os.remove(shape_file)
+        return zip_filepath
+
+    def export_layer(
+        self, type: FileType, working_dir: str, filename = None
+    ):
+        """
+        Export the current layer to requested format.
+
+        This method converts a layer in postgis table into
+        shapefile/geojson/GPKG/KML using the 'ogr2ogr.
+
+        Returns:
+            tuple:
+                - bool: Success status of the operation.
+                - str: Message indicating the outcome
+        """
+        driver_dict = {
+            FileType.GEOJSON: 'GeoJSON',
+            FileType.GEOPACKAGE: 'GPKG',
+            FileType.KML: 'KML',
+            FileType.SHAPEFILE: 'ESRI Shapefile'
+        }
+        ext = (
+            '.shp' if type == FileType.SHAPEFILE else
+            FileType.to_extension(type)
+        )
+        name = Path(filename).stem if filename else str(self.unique_id)
+        export_filepath = os.path.join(
+            working_dir,
+            f'{name}{ext}'
+        )
+        conn_str = (
+            'PG:dbname={NAME} user={USER} password={PASSWORD} '
+            'host={HOST} port={PORT}'.format(
+                **connection.settings_dict
+            )
+        )
+        sql_str = (
+            'SELECT * FROM {table_name}'.format(
+                table_name=self.query_table_name
+            )
+        )
+        cmd_list = [
+            'ogr2ogr',
+            '-t_srs',
+            'EPSG:4326',
+            '-f',
+            f'{driver_dict[type]}',
+            export_filepath,
+            conn_str,
+            '-sql',
+            sql_str
+        ]
+        if type == FileType.SHAPEFILE:
+            cmd_list.append('-lco')
+            cmd_list.append('ENCODING=UTF-8')
+
+        try:
+            subprocess.run(cmd_list, check=True)
+
+            if type == FileType.SHAPEFILE:
+                # zip the files
+                export_filepath = self._zip_shapefile(
+                    export_filepath, working_dir
+                )
+
+            return (
+                export_filepath,
+                'Success'
+            )
+        except subprocess.CalledProcessError:
+            return (
+                None,
+                f'Failed to export layer {self.name} to format {type}'
+            )
+
 
 class LayerAttributes(models.Model):
     """Field of layer."""

django_project/cloud_native_gis/models/layer_upload.py

@@ -19,7 +19,7 @@
 from cloud_native_gis.utils.connection import fields
 from cloud_native_gis.utils.geopandas import collection_to_postgis
 from cloud_native_gis.utils.main import id_generator
-from cloud_native_gis.utils.fiona import FileType
+from cloud_native_gis.utils.type import FileType
 
 FOLDER_FILES = 'cloud_native_gis_files'
 FOLDER_ROOT = os.path.join(

django_project/cloud_native_gis/tests/models/__init__.py

@@ -0,0 +1 @@
+from .layer import *

django_project/cloud_native_gis/tests/models/layer.py

@@ -0,0 +1,76 @@
+# coding=utf-8
+"""Cloud Native GIS."""
+
+import os
+import uuid
+from django.test import TestCase
+from django.conf import settings
+import shutil
+
+from cloud_native_gis.models import (
+    Layer, LayerUpload,
+    UploadStatus
+)
+from cloud_native_gis.tests.model_factories import create_user
+from cloud_native_gis.utils.connection import count_features
+from cloud_native_gis.utils.main import ABS_PATH
+from cloud_native_gis.utils.type import FileType
+
+
+class TestLayerModel(TestCase):
+    """Test class for layer models."""
+
+    def setUp(self):
+        """To setup test."""
+        self.user = create_user()
+
+    def test_export_layer(self):
+        filepath = ABS_PATH(
+            'cloud_native_gis', 'tests', '_fixtures',
+            'capital_cities.zip'
+        )
+
+        layer = Layer.objects.create(
+            unique_id=uuid.uuid4(),
+            name='Test Layer',
+            created_by=self.user
+        )
+        layer_upload = LayerUpload.objects.create(
+            layer=layer,
+            created_by=self.user
+        )
+        # copy to folder data
+        layer_upload.emptying_folder()
+        shutil.copy(filepath, layer_upload.folder)
+        # import shapefile
+        layer_upload.import_data()
+
+        # assert layer
+        layer.refresh_from_db()
+        layer_upload.refresh_from_db()
+
+        self.assertEqual(layer_upload.status, UploadStatus.SUCCESS)
+        self.assertEqual(
+            layer.attribute_names, ['CITY_NAME', 'CITY_TYPE', 'COUNTRY']
+        )
+        # Check count features
+        self.assertEqual(
+            count_features(layer.schema_name, layer.table_name),
+            layer.metadata['FEATURE COUNT']
+        )
+
+        # try export
+        export_path, msg = layer.export_layer(
+            FileType.SHAPEFILE, '/tmp', f'{str(layer.unique_id)}.shp'
+        )
+        self.assertIsNotNone(export_path)
+        self.assertEqual(
+            export_path,
+            os.path.join('/tmp', f'{str(layer.unique_id)}.zip')
+        )
+        self.assertEqual(msg, 'Success')
+
+        os.remove(export_path)
+
+        layer.delete()
+        self.assertFalse(os.path.exists(layer_upload.folder))

django_project/cloud_native_gis/tests/utils/fiona.py

@@ -9,8 +9,8 @@
 )
 
 from core.settings.utils import absolute_path
+from cloud_native_gis.utils.type import FileType
 from cloud_native_gis.utils.fiona import (
-    FileType,
     validate_shapefile_zip,
     open_fiona_collection,
     validate_collection_crs,

django_project/cloud_native_gis/utils/fiona.py

@@ -12,47 +12,13 @@
     TemporaryUploadedFile
 )
 
+from cloud_native_gis.utils.type import FileType
+
 
 # Enable KML driver support in Fiona
 fiona.drvsupport.supported_drivers['KML'] = 'rw'
 
 
-class FileType:
-    """File types."""
-
-    GEOJSON = 'geojson'
-    SHAPEFILE = 'shapefile'
-    GEOPACKAGE = 'geopackage'
-    KML = 'kml'
-
-    @staticmethod
-    def guess_type(filename: str):
-        """Guess file type based on filename."""
-        if filename.endswith('.geojson') or filename.endswith('.json'):
-            return FileType.GEOJSON
-        elif filename.endswith('.zip') or filename.endswith('.shp'):
-            return FileType.SHAPEFILE
-        elif filename.endswith('.gpkg'):
-            return FileType.GEOPACKAGE
-        elif filename.endswith('.kml'):
-            return FileType.KML
-
-        return None
-
-    @staticmethod
-    def to_extension(type: str):
-        """Convert FileType to file extension."""
-        if type == FileType.GEOJSON:
-            return '.geojson'
-        elif type == FileType.SHAPEFILE:
-            return '.zip'
-        elif type == FileType.GEOPACKAGE:
-            return '.gpkg'
-        elif type == FileType.KML:
-            return '.kml'
-        return ''
-
-
 def _open_collection(fp: str, type: str) -> Collection:
     """Open collection from file path."""
     if type == FileType.SHAPEFILE:

django_project/cloud_native_gis/utils/type.py

@@ -0,0 +1,38 @@
+# coding=utf-8
+"""Cloud Native GIS."""
+
+
+class FileType:
+    """File types."""
+
+    GEOJSON = 'geojson'
+    SHAPEFILE = 'shapefile'
+    GEOPACKAGE = 'geopackage'
+    KML = 'kml'
+
+    @staticmethod
+    def guess_type(filename: str):
+        """Guess file type based on filename."""
+        if filename.endswith('.geojson') or filename.endswith('.json'):
+            return FileType.GEOJSON
+        elif filename.endswith('.zip') or filename.endswith('.shp'):
+            return FileType.SHAPEFILE
+        elif filename.endswith('.gpkg'):
+            return FileType.GEOPACKAGE
+        elif filename.endswith('.kml'):
+            return FileType.KML
+
+        return None
+
+    @staticmethod
+    def to_extension(type: str):
+        """Convert FileType to file extension."""
+        if type == FileType.GEOJSON:
+            return '.geojson'
+        elif type == FileType.SHAPEFILE:
+            return '.zip'
+        elif type == FileType.GEOPACKAGE:
+            return '.gpkg'
+        elif type == FileType.KML:
+            return '.kml'
+        return ''