Fix/pq dataset typology

digital_land/package/datasetparquet.py

@@ -1,6 +1,5 @@
 import os
 import logging
-from pathlib import Path
 import duckdb
 from .package import Package
 
@@ -37,6 +36,7 @@ def __init__(self, dataset, input_paths, **kwargs):
         self.duckdb_file = "input_paths_database.duckdb"
         self.conn = duckdb.connect(self.duckdb_file)
         self.schema = self.get_schema(input_paths)
+        self.typology = self.specification.schema[dataset]["typology"]
 
     def get_schema(self, input_paths):
         # There are issues with the schema when reading in lots of files, namely smaller files have few or zero rows
@@ -149,7 +149,12 @@ def load_entities(
 
         # json fields - list of fields which are present in the fact table which
         # do not exist separately in the entity table
-        json_fields = [field for field in distinct_fields if field not in entity_fields]
+        # Need to ensure that 'organisation' is not included either
+        json_fields = [
+            field
+            for field in distinct_fields
+            if field not in entity_fields + ["organisation"]
+        ]
 
         # null fields - list of fields which are not present in the fact tables which have
         # to be in the entity table as a column
@@ -207,8 +212,13 @@ def load_entities(
         # include columns in the json statement
         # Collate list of fields which don't exist but need to be in the final table
         select_statement = ", ".join([f"t1.{field}" for field in select_fields])
+        # Don't want to include anything that ends with "_geom"
         null_fields_statement = ", ".join(
-            [f'NULL::VARCHAR AS "{field}"' for field in null_fields]
+            [
+                f'NULL::VARCHAR AS "{field}"'
+                for field in null_fields
+                if not field.endswith("_geom")
+            ]
         )
         json_statement = ", ".join(
             [
@@ -223,7 +233,6 @@ def load_entities(
              SELECT * FROM read_csv_auto('{org_csv}')
          """
 
-        dataset = Path(output_path).name
         sql = f"""
             INSTALL spatial; LOAD spatial;
             COPY(
@@ -236,8 +245,8 @@ def load_entities(
                             ELSE point
                         END AS point
                     FROM (
-                        SELECT '{dataset}' as dataset,
-                        '{dataset}' as typology,
+                        SELECT '{self.dataset}' as dataset,
+                        '{self.typology}' as typology,
                         t2.entity as organisation_entity,
                         {select_statement},
                         {null_fields_statement},

tests/integration/test_package_datasetparquet.py

@@ -199,8 +199,6 @@ def test_dataset_parquet_package(temp_dir):
         for row in data:
             f.write(",".join(map(str, row)) + "\n")
 
-    # df = pd.read_csv(input_paths[0])
-
     # Test data for the tables. This has plenty of 'duplicates' to check
     data = [
         [
@@ -462,7 +460,7 @@ def test_dataset_parquet_package(temp_dir):
 
     # Instantiate the DatasetParquetPackage with temp_dir input paths and a mock schema
     package = DatasetParquetPackage(
-        dataset="test_dataset", input_paths=input_paths, specification_dir=None
+        dataset="conservation-area", input_paths=input_paths, specification_dir=None
     )
     package.create_temp_table(input_paths)
 
@@ -534,11 +532,9 @@ def test_load_entities_basic(test_dataset_parquet_package, temp_dir):
     df = pd.read_parquet(output_file)
     assert len(df) > 0, "No data in entity.parquet file"
     assert len(df) == 11, "No. of entities is not correct"
-    assert df.shape[1] == 16, "Not all columns saved in entity.parquet file"
+    assert df.shape[1] == 14, "Not all columns saved in entity.parquet file"
     assert df["end_date"].isnull().all()  # Check null handling
     assert df["geojson"].isnull().all()  # Check null handling
-    assert df["geometry_geom"].isnull().all()  # Check null handling
-    assert df["point_geom"].isnull().all()  # Check null handling
 
 
 def test_load_pq_to_sqlite_basic(test_dataset_parquet_package, temp_dir):