Bugfix/18 fix kartverket api error on cloud

src/application/contracts/county_service_interface.py

@@ -10,5 +10,5 @@ def get_county_ids(self) -> list[str]:
         raise NotImplementedError
 
     @abstractmethod
-    def get_county_wkb_by_id(self, county_id: str, epsg_code: EPSGCode) -> tuple[bytes, dict[str, Any]]:
+    def get_county_polygons_by_id(self, county_id: str, epsg_code: EPSGCode) -> tuple[bytes, dict[str, Any]]:
         raise NotImplementedError

src/config.py

@@ -50,7 +50,8 @@ class Config:
     GEONORGE_BASE_URL: str = "https://api.kartverket.no/kommuneinfo/v1/"
 
     # METADATA
-    RELEASE_FILE_NAME = "releases.parquet"
+    RELEASE_FILE_NAME: str = "releases.parquet"
+    COUNTY_FILE_NAME: str = "counties.parquet"
 
     # STAC
     STAC_LICENSE = "CC-BY-4.0"

src/infra/infrastructure/containers.py

@@ -13,11 +13,6 @@ class Containers(containers.DeclarativeContainer):
     db_context = providers.Singleton(create_duckdb_context)
     blob_storage_context = providers.Singleton(create_blob_storage_context)
 
-    county_service = providers.Singleton(
-        CountyService,
-        db_context=db_context
-    )
-
     file_path_service = providers.Singleton(
         FilePathService
     )
@@ -37,6 +32,13 @@ class Containers(containers.DeclarativeContainer):
         file_path_service=file_path_service
     )
 
+    county_service = providers.Singleton(
+        CountyService,
+        db_context=db_context,
+        blob_storage_service=blob_storage_service,
+        bytes_service=bytes_service
+    )
+
     osm_file_service = providers.Singleton(
         OpenStreetMapFileService
     )

src/infra/infrastructure/services/county_service.py

@@ -1,41 +1,109 @@
-from typing import Any
+import json
+from io import BytesIO
+from typing import Any, Dict
 
+import pandas as pd
 import requests
 import shapely
 from duckdb import DuckDBPyConnection
 from shapely.geometry import shape
 
 from src import Config
-from src.application.contracts import ICountyService
-from src.domain.enums import EPSGCode
+from src.application.contracts import ICountyService, IBlobStorageService, IBytesService
+from src.domain.enums import EPSGCode, StorageContainer
 
 
 class CountyService(ICountyService):
     __db_context: DuckDBPyConnection
+    __blob_storage_service: IBlobStorageService
+    __bytes_service: IBytesService
 
-    def __init__(self, db_context: DuckDBPyConnection):
+    def __init__(
+            self,
+            db_context: DuckDBPyConnection,
+            blob_storage_service: IBlobStorageService,
+            bytes_service: IBytesService
+    ):
         self.__db_context = db_context
+        self.__blob_storage_service = blob_storage_service
+        self.__bytes_service = bytes_service
 
     def get_county_ids(self) -> list[str]:
         response = requests.get(f"{Config.GEONORGE_BASE_URL}/fylker?sorter=fylkesnummer")
         response.raise_for_status()
         data = response.json()
         return [item["fylkesnummer"] for item in data]
 
-    def get_county_wkb_by_id(self, county_id: str, epsg_code: EPSGCode) -> tuple[bytes, dict[str, Any]]:
+    def get_county_polygons_by_id(self, county_id: str, epsg_code: EPSGCode) -> tuple[bytes, dict[str, Any]]:
+        geometries = self.__get_county_polygons_from_blob_storage(region=county_id)
+
+        if geometries is not None:
+            return geometries
+
+        wkb, geo_json = self.__fetch_county_polygons_from_api(region=county_id, epsg_code=epsg_code)
+        self.__write_county_to_blob_storage(region=county_id, wkb=wkb, geo_json=geo_json)
+
+        return wkb, geo_json
+
+    @staticmethod
+    def __fetch_county_polygons_from_api(region: str, epsg_code: EPSGCode) -> tuple[bytes, dict[str, Any]]:
         response = requests.get(
-            f"{Config.GEONORGE_BASE_URL}/fylker/{county_id}/omrade?utkoordsys={epsg_code.value}"
+            f"{Config.GEONORGE_BASE_URL}/fylker/{region}/omrade?utkoordsys={epsg_code.value}"
         )
         response.raise_for_status()
 
         data = response.json()
         geom_data = data["omrade"]
         geom = shape(geom_data)
-        wkb_data = shapely.to_wkb(geom)
+        wkb = shapely.to_wkb(geom)
 
         geo_json = {
             "type": geom_data["type"],
             "coordinates": geom_data["coordinates"]
         }
 
-        return wkb_data, geo_json
+        return wkb, geo_json
+
+    def __get_county_polygons_from_blob_storage(self, region: str) -> tuple[bytes, dict[str, Any]] | None:
+        county_bytes = self.__blob_storage_service.download_file(
+            container_name=StorageContainer.METADATA,
+            blob_name=Config.COUNTY_FILE_NAME
+        )
+
+        if county_bytes is None:
+            return None
+
+        county_df = self.__bytes_service.convert_parquet_bytes_to_df(county_bytes)
+        if region not in county_df["region"].values:
+            return None
+
+        row = county_df.loc[county_df["region"] == region].iloc[0]
+        return row["wkb"], json.loads(row["json"])
+
+    def __write_county_to_blob_storage(self, region: str, wkb: bytes, geo_json: Dict[str, Any]) -> None:
+        county_bytes = self.__blob_storage_service.download_file(
+            container_name=StorageContainer.METADATA,
+            blob_name=Config.COUNTY_FILE_NAME
+        )
+
+        county_df = self.__bytes_service.convert_parquet_bytes_to_df(county_bytes) \
+            if (county_bytes is not None and len(county_bytes) > 0) \
+            else pd.DataFrame(columns=["region", "wkb", "json"])
+
+        json_string = json.dumps(geo_json)
+        if region in county_df["region"].values:
+            mask = county_df["region"] == region
+            county_df.loc[mask, "wkb"] = wkb
+            county_df.loc[mask, "json"] = json_string
+        else:
+            new_row = pd.DataFrame({"region": [region], "wkb": [wkb], "json": [json_string]})
+            county_df = pd.concat([county_df, new_row], ignore_index=True)
+
+        buffer = BytesIO()
+        county_df.to_parquet(buffer, index=False)
+        buffer.seek(0)
+        self.__blob_storage_service.upload_file(
+            container_name=StorageContainer.METADATA,
+            blob_name=Config.COUNTY_FILE_NAME,
+            data=buffer.read()
+        )

src/presentation/entrypoints/release_pipeline.py

@@ -157,7 +157,7 @@ def clip_and_partition_dataset_to_region(
         county_service: ICountyService = Provide[Containers.county_service],
         vector_service: IVectorService = Provide[Containers.vector_service],
 ) -> tuple[list[gpd.GeoDataFrame], list[gpd.GeoDataFrame], dict[str, Any]]:
-    polygon_wkb, polygon_geojson = county_service.get_county_wkb_by_id(county_id=region, epsg_code=EPSGCode.WGS84)
+    polygon_wkb, polygon_geojson = county_service.get_county_polygons_by_id(county_id=region, epsg_code=EPSGCode.WGS84)
 
     osm_county_dataset = vector_service.clip_dataframes_to_wkb(osm_batches, polygon_wkb, epsg_code=EPSGCode.WGS84)
     fkb_county_dataset = vector_service.clip_dataframes_to_wkb(fkb_batches, polygon_wkb, epsg_code=EPSGCode.WGS84)