Merge pull request #19 from PDOK/syntax-fix

fix minor syntax issues

ngr_spider/ogc_api_tiles.py

@@ -5,20 +5,20 @@
 
 LOGGER = logging.getLogger(__name__)
 
+
 class Info:
     description: str
     title: str
     version: str
 
-    def __init__(self, data: []):
+    def __init__(self, data: dict):
         self.description = data["description"]
         self.title = data["title"]
         self.version = data["version"]
         pass
 
 
 class ServiceDesc:
-    json: []
     def __init__(self, href: str):
         with urllib.request.urlopen(href) as url:
             self.json = json.load(url)
@@ -32,33 +32,38 @@ def get_tags(self):
     def get_servers(self):
         return self.json["servers"]
 
-    def __get_url_from_servers(self, servers: []):
+    def __get_url_from_servers(self, servers: list[str]):
         for server in servers:
             if len(server["url"]) > 0:
                 return server["url"]
 
     def get_tile_request_url(self):
         paths = self.json["paths"]
         for path in paths:
-            if "{tileMatrixSetId}" in path and "{tileMatrix}" in path and "{tileRow}" in path and "{tileCol}" in path:
-                return self.__get_url_from_servers(self.get_servers()) + path # kijken in de spec of service url altijd goed is
+            if (
+                "{tileMatrixSetId}" in path
+                and "{tileMatrix}" in path
+                and "{tileRow}" in path
+                and "{tileCol}" in path
+            ):
+                return (
+                    self.__get_url_from_servers(self.get_servers()) + path
+                )  # kijken in de spec of service url altijd goed is
+
 
 class Data:
-    json: []
     def __init__(self, href: str):
         with urllib.request.urlopen(href) as url:
             self.json = json.load(url)
 
 
 class Tiles:
-    json: []
     def __init__(self, href: str):
         with urllib.request.urlopen(href) as url:
             self.json = json.load(url)
 
 
 class TileMatrixSets:
-    json: []
     def __init__(self, href: str):
         with urllib.request.urlopen(href) as url:
             self.json = json.load(url)
@@ -93,7 +98,7 @@ def get_layers(self):
 
         # process styles
         # TODO style should be generated based on the type of the tiles; png, Vector etc.
-        vector_tile_styles: [VectorTileStyle] = self.get_styles()
+        vector_tile_styles: list[VectorTileStyle] = self.get_styles()
 
         # process min/max resolution
         layer_tile_set_matrix = self.get_tile_matrix_sets()
@@ -109,9 +114,12 @@ def get_layers(self):
             layer_tile_matrix_set_id = tile_set["tileMatrixSetId"]
             if layer_tile_matrix_set_id in layer_tile_set_matrix:
                 # https://docs.kadaster.nl/ggc/ggs-ggc-library/algemeen/scale-set/
-                service_layer_min_scale = layer_tile_set_matrix[layer_tile_matrix_set_id]
+                service_layer_min_scale = layer_tile_set_matrix[
+                    layer_tile_matrix_set_id
+                ]
             service_layer_crs = tile_set["crs"]
-            service_layer_title = tile_set["title"]
+
+            service_layer_title = tile_set["title"] if "title" in tile_set else ""
             t_links = tile_set["links"]
             for l in t_links:
                 if l["rel"] == "self":
@@ -120,18 +128,25 @@ def get_layers(self):
                         service_layer_title = tile["title"]
                         self.service_type = tile["dataType"]
 
-        return [OatLayer(service_layer_name,
-                         service_layer_title,
-                         service_layer_abstract,
-                         "",
-                         vector_tile_styles,
-                         service_layer_crs,
-                         service_layer_min_scale,
-                         service_layer_max_scale)]
+        return [
+            OatLayer(
+                service_layer_name,
+                service_layer_title,
+                service_layer_abstract,
+                "",
+                vector_tile_styles,
+                service_layer_crs,
+                service_layer_min_scale,
+                service_layer_max_scale,
+            )
+        ]
 
     def __load_landing_page(self, service_url: str):
-        with urllib.request.urlopen(service_url) as url:
-            links = json.load(url)["links"]
+        with urllib.request.urlopen(service_url) as response:
+            response_body = response.read().decode("utf-8")
+            response_body_data = json.loads(response_body)
+
+            links = response_body_data["links"]
             for link in links:
                 if link["rel"] == "service-desc":
                     self.service_desc = ServiceDesc(link["href"])
@@ -141,11 +156,13 @@ def __load_landing_page(self, service_url: str):
                     self.tiles = Tiles(link["href"])
                 elif link["rel"] == "tileMatrixSets":
                     self.tile_matrix_sets = TileMatrixSets(link["href"])
-            self.title = json.load(url)["title"] if json.load(url)["title"] is not None else ""
-            self.description = json.load(url)["description"] if json.load(url)["description"] is not None else ""
+            title = response_body_data["title"]
+            self.title = title if title else ""
+            description = response_body_data["description"]
+            self.description = description if description else ""
 
     def get_styles(self):
-        styles: [VectorTileStyle] = []
+        styles: list[VectorTileStyle] = []
         data = self.data.json
         default_style_name: str = ""
         if data["default"] is not None:

ngr_spider/util.py

@@ -264,8 +264,9 @@ def get_atom_service(
     r = requests.get(service_record.service_url)
     return AtomService(service_record.service_url, r.text)
 
+
 def get_oat_service(
-        service_record: CswServiceRecord,
+    service_record: CswServiceRecord,
 ) -> Union[OatService, ServiceError]:
     try:
         url = service_record.service_url
@@ -276,18 +277,23 @@ def get_oat_service(
             return service_record
         oat = OGCApiTiles(url)
         title = oat.title
-        if title is "": # fallback
+        if title == "":  # fallback
             title = empty_string_if_none(oat.service_desc.get_info().title)
         description = oat.description
-        if description is "": # fallback
+        if description == "":  # fallback
             description = empty_string_if_none(oat.service_desc.get_info().description)
+
+        layers = oat.get_layers()
+        for layer in layers:
+            layer.dataset_metadata_id = service_record.dataset_metadata_id
+
         return OatService(
             # http://docs.ogc.org/DRAFTS/19-072.html#rc_landing-page-section
             title=title,
             abstract=description,
             metadata_id=md_id,
             url=oat.service_desc.get_tile_request_url(),
-            layers=oat.get_layers(),
+            layers=layers,
             keywords=oat.service_desc.get_tags(),
             dataset_metadata_id=service_record.dataset_metadata_id,
         )
@@ -298,6 +304,7 @@ def get_oat_service(
         LOGGER.exception(message)
     return ServiceError(service_record.service_url, service_record.metadata_id)
 
+
 def get_wms_service(
     service_record: CswServiceRecord,
 ) -> Union[WmsService, ServiceError]: