Fix missing types to core pygeoapi (#2143)

* Fix API types

* Fix more signatures

Author: francbartoli

pygeoapi/api/__init__.py

@@ -49,8 +49,9 @@
 import logging
 import re
 import sys
-from typing import Any, Tuple, Union, Optional
+from typing import Any, Tuple, Union, Optional, Self
 
+from babel import Locale
 from dateutil.parser import parse as dateparse
 import pytz
 
@@ -286,7 +287,8 @@ def _get_params(request):
         LOGGER.debug('No query parameters found')
         return {}
 
-    def _get_locale(self, headers, supported_locales):
+    def _get_locale(self, headers: dict,
+                    supported_locales: Union[list, Locale]) -> tuple:
         """
         Detects locale from "lang=<language>" param or `Accept-Language`
         header. Returns a tuple of (raw, locale) if found in params or headers.
@@ -325,7 +327,7 @@ def _get_locale(self, headers, supported_locales):
 
         return raw, default_locale
 
-    def _get_format(self, headers) -> Union[str, None]:
+    def _get_format(self, headers: dict) -> Union[str, None]:
         """
         Get `Request` format type from query parameters or headers.
 
@@ -464,9 +466,9 @@ def is_valid(self, additional_formats=None) -> bool:
             return True
         return False
 
-    def get_response_headers(self, force_lang: l10n.Locale = None,
-                             force_type: str = None,
-                             force_encoding: str = None,
+    def get_response_headers(self, force_lang: l10n.Locale | None = None,
+                             force_type: str | None = None,
+                             force_encoding: str | None = None,
                              **custom_headers) -> dict:
         """
         Prepares and returns a dictionary with Response object headers.
@@ -512,7 +514,7 @@ def get_response_headers(self, force_lang: l10n.Locale = None,
 
         return headers
 
-    def get_request_headers(self, headers) -> dict:
+    def get_request_headers(self, headers: dict) -> dict:
         """
         Obtains and returns a dictionary with Request object headers.
 
@@ -529,7 +531,7 @@ def get_request_headers(self, headers) -> dict:
 class API:
     """API object"""
 
-    def __init__(self, config, openapi):
+    def __init__(self, config: dict, openapi: dict) -> Self | None:
         """
         constructor
 
@@ -571,8 +573,8 @@ def __init__(self, config, openapi):
         self.manager = get_manager(self.config)
         LOGGER.info('Process manager plugin loaded')
 
-    def get_exception(self, status, headers, format_, code,
-                      description) -> Tuple[dict, int, str]:
+    def get_exception(self, status: int, headers: dict, format_: str | None,
+                      code: str, description: str) -> Tuple[dict, int, str]:
         """
         Exception handler
 
@@ -610,7 +612,8 @@ def get_exception(self, status, headers, format_, code,
 
         return headers, status, content
 
-    def get_format_exception(self, request) -> Tuple[dict, int, str]:
+    def get_format_exception(self,
+                             request: APIRequest) -> Tuple[dict, int, str]:
         """
         Returns a format exception.
 
@@ -631,7 +634,7 @@ def get_format_exception(self, request) -> Tuple[dict, int, str]:
     def get_collections_url(self) -> str:
         return f"{self.base_url}/collections"
 
-    def get_dataset_templates(self, dataset) -> dict:
+    def get_dataset_templates(self, dataset: str) -> dict:
         templates = self.config['resources'][dataset].get('templates')
 
         return templates or self.tpl_config['server']['templates']
@@ -884,7 +887,7 @@ def openapi_(api: API, request: APIRequest) -> Tuple[dict, int, str]:
         return headers, HTTPStatus.OK, api.openapi
 
 
-def conformance(api, request: APIRequest) -> Tuple[dict, int, str]:
+def conformance(api: API, request: APIRequest) -> Tuple[dict, int, str]:
     """
     Provide conformance definition
 
@@ -930,7 +933,7 @@ def conformance(api, request: APIRequest) -> Tuple[dict, int, str]:
 
 @jsonldify
 def describe_collections(api: API, request: APIRequest,
-                         dataset=None) -> Tuple[dict, int, str]:
+                         dataset: str | None = None) -> Tuple[dict, int, str]:
     """
     Provide collection metadata
 
@@ -1377,7 +1380,7 @@ def describe_collections(api: API, request: APIRequest,
 
 
 def get_collection_schema(api: API, request: Union[APIRequest, Any],
-                          dataset) -> Tuple[dict, int, str]:
+                          dataset: str) -> Tuple[dict, int, str]:
     """
     Returns a collection schema
 
@@ -1462,7 +1465,7 @@ def get_collection_schema(api: API, request: Union[APIRequest, Any],
     return headers, HTTPStatus.OK, to_json(schema, api.pretty_print)
 
 
-def validate_bbox(value=None) -> list:
+def validate_bbox(value: list | None = None) -> list:
     """
     Helper function to validate bbox parameter
 
@@ -1510,7 +1513,8 @@ def validate_bbox(value=None) -> list:
     return bbox
 
 
-def validate_datetime(resource_def, datetime_=None) -> str:
+def validate_datetime(resource_def: dict,
+                      datetime_: str | None = None) -> str:
     """
     Helper function to validate temporal parameter
 

pygeoapi/api/admin.py

@@ -54,7 +54,7 @@ class Admin(API):
     PYGEOAPI_CONFIG = os.environ.get('PYGEOAPI_CONFIG')
     PYGEOAPI_OPENAPI = os.environ.get('PYGEOAPI_OPENAPI')
 
-    def __init__(self, config, openapi):
+    def __init__(self, config: dict, openapi: dict):
         """
         constructor
 
@@ -66,7 +66,7 @@ def __init__(self, config, openapi):
 
         super().__init__(config, openapi)
 
-    def merge(self, obj1, obj2):
+    def merge(self, obj1: dict, obj2: dict):
         """
         Merge two dictionaries
 
@@ -89,7 +89,7 @@ def merge(self, obj1, obj2):
         else:
             return obj2
 
-    def validate(self, config):
+    def validate(self, config: dict):
         """
         Validate pygeoapi configuration and OpenAPI to file
 
@@ -105,7 +105,7 @@ def validate(self, config):
         # validate_openapi_document(oas)
         return True
 
-    def write(self, config):
+    def write(self, config: dict):
         """
         Write pygeoapi configuration and OpenAPI to file
 
@@ -115,7 +115,7 @@ def write(self, config):
         self.write_config(config)
         self.write_oas(config)
 
-    def write_config(self, config):
+    def write_config(self, config: dict):
         """
         Write pygeoapi configuration file
 
@@ -140,7 +140,7 @@ def write_config(self, config):
         yaml_dump(config, self.PYGEOAPI_CONFIG)
         LOGGER.debug('Finished writing pygeoapi configuration')
 
-    def write_oas(self, config):
+    def write_oas(self, config: dict):
         """
         Write pygeoapi OpenAPI document
 

pygeoapi/api/coverages.py

@@ -69,7 +69,8 @@
 
 
 def get_collection_coverage(
-        api: API, request: APIRequest, dataset) -> Tuple[dict, int, str]:
+        api: API, request: APIRequest,
+        dataset: str) -> Tuple[dict, int, str]:
     """
     Returns a subset of a collection coverage
 

pygeoapi/api/environmental_data_retrieval.py

@@ -66,8 +66,10 @@
 ]
 
 
-def get_collection_edr_instances(api: API, request: APIRequest, dataset,
-                                 instance_id=None) -> Tuple[dict, int, str]:
+def get_collection_edr_instances(api: API, request: APIRequest,
+                                 dataset: str,
+                                 instance_id: str | None = None
+                                 ) -> Tuple[dict, int, str]:
     """
     Queries collection EDR instances
 
@@ -233,8 +235,10 @@ def get_collection_edr_instances(api: API, request: APIRequest, dataset,
 
 
 def get_collection_edr_query(api: API, request: APIRequest,
-                             dataset, instance, query_type,
-                             location_id=None) -> Tuple[dict, int, str]:
+                             dataset: str, instance: str,
+                             query_type: str,
+                             location_id: str | None = None
+                             ) -> Tuple[dict, int, str]:
     """
     Queries collection EDR
 

pygeoapi/api/itemtypes.py

@@ -101,7 +101,8 @@
 
 
 def get_collection_queryables(api: API, request: Union[APIRequest, Any],
-                              dataset=None) -> Tuple[dict, int, str]:
+                              dataset: str | None = None
+                              ) -> Tuple[dict, int, str]:
     """
     Provide collection queryables
 
@@ -235,7 +236,7 @@ def get_collection_queryables(api: API, request: Union[APIRequest, Any],
 
 def get_collection_items(
         api: API, request: Union[APIRequest, Any],
-        dataset) -> Tuple[dict, int, str]:
+        dataset: str | None = None) -> Tuple[dict, int, str]:
     """
     Queries collection
 
@@ -703,7 +704,8 @@ def get_collection_items(
 
 def manage_collection_item(
         api: API, request: APIRequest,
-        action, dataset, identifier=None) -> Tuple[dict, int, str]:
+        action: str, dataset: str,
+        identifier: str | None = None) -> Tuple[dict, int, str]:
     """
     Adds an item to a collection
 
@@ -813,7 +815,8 @@ def manage_collection_item(
 
 
 def get_collection_item(api: API, request: APIRequest,
-                        dataset, identifier) -> Tuple[dict, int, str]:
+                        dataset: str, identifier: str
+                        ) -> Tuple[dict, int, str]:
     """
     Get a single collection item
 
@@ -986,7 +989,8 @@ def get_collection_item(api: API, request: APIRequest,
 
 
 def create_crs_transform_spec(
-        config: dict, query_crs_uri: Optional[str] = None) -> Union[None, CrsTransformSpec]:  # noqa
+        config: dict,
+        query_crs_uri: Optional[str] = None) -> Union[None, CrsTransformSpec]:  # noqa
     """
     Create a `CrsTransformSpec` instance based on provider config and
     *crs* query parameter.

pygeoapi/api/maps.py

@@ -64,7 +64,8 @@
 
 
 def get_collection_map(api: API, request: APIRequest,
-                       dataset, style=None) -> Tuple[dict, int, str]:
+                       dataset: str, style: str | None = None
+                       ) -> Tuple[dict, int, str]:
     """
     Returns a subset of a collection map
 
@@ -194,7 +195,8 @@ def get_collection_map(api: API, request: APIRequest,
 
 
 def get_collection_map_legend(api: API, request: APIRequest,
-                              dataset, style=None) -> Tuple[dict, int, str]:
+                              dataset: str, style: str | None = None
+                              ) -> Tuple[dict, int, str]:
     """
     Returns a subset of a collection map legend
 

pygeoapi/api/processes.py

@@ -40,18 +40,20 @@
 # =================================================================
 
 
+import json
+import logging
+import urllib.parse
 from copy import deepcopy
 from datetime import datetime, timezone
 from http import HTTPStatus
-import json
-import logging
 from typing import Tuple
-import urllib.parse
 
 from pygeoapi import l10n
 from pygeoapi.api import evaluate_limit
 from pygeoapi.process.base import (
-    JobNotFoundError, JobResultNotFoundError, ProcessorExecuteError
+    JobNotFoundError,
+    JobResultNotFoundError,
+    ProcessorExecuteError,
 )
 from pygeoapi.process.manager.base import get_manager, Subscriber
 from pygeoapi.util import (
@@ -74,7 +76,8 @@
 
 
 def describe_processes(api: API, request: APIRequest,
-                       process=None) -> Tuple[dict, int, str]:
+                       process: str | None = None
+                       ) -> Tuple[dict, int, str]:
     """
     Provide processes metadata
 
@@ -227,7 +230,7 @@ def describe_processes(api: API, request: APIRequest,
 
 # TODO: get_jobs doesn't have tests
 def get_jobs(api: API, request: APIRequest,
-             job_id=None) -> Tuple[dict, int, str]:
+             job_id: str | None = None) -> Tuple[dict, int, str]:
     """
     Get process jobs
 
@@ -405,7 +408,7 @@ def get_jobs(api: API, request: APIRequest,
 
 
 def execute_process(api: API, request: APIRequest,
-                    process_id) -> Tuple[dict, int, str]:
+                    process_id: str) -> Tuple[dict, int, str]:
     """
     Execute process
 
@@ -532,7 +535,7 @@ def execute_process(api: API, request: APIRequest,
 
 
 def get_job_result(api: API, request: APIRequest,
-                   job_id) -> Tuple[dict, int, str]:
+                   job_id: str) -> Tuple[dict, int, str]:
     """
     Get result of job (instance of a process)
 
@@ -602,7 +605,8 @@ def get_job_result(api: API, request: APIRequest,
     return headers, HTTPStatus.OK, content
 
 
-def delete_job(api: API, request: APIRequest, job_id) -> Tuple[dict, int, str]:
+def delete_job(api: API, request: APIRequest,
+               job_id: str) -> Tuple[dict, int, str]:
     """
     Delete a process job
 
@@ -647,7 +651,8 @@ def delete_job(api: API, request: APIRequest, job_id) -> Tuple[dict, int, str]:
     return {}, http_status, to_json(response, api.pretty_print)
 
 
-def get_oas_30(cfg: dict, locale: str) -> tuple[list[dict[str, str]], dict[str, dict]]:  # noqa
+def get_oas_30(cfg: dict, locale: str
+               ) -> tuple[list[dict[str, str]], dict[str, dict]]:  # noqa
     """
     Get OpenAPI fragments
 

pygeoapi/api/stac.py

@@ -134,7 +134,7 @@ def get_stac_root(api: API, request: APIRequest) -> Tuple[dict, int, str]:
 
 # TODO: no tests for this?
 def get_stac_path(api: API, request: APIRequest,
-                  path) -> Tuple[dict, int, str]:
+                  path: str) -> Tuple[dict, int, str]:
     """
     Provide STAC resource path