Skip to content

Commit

Permalink
refactor: aws auth typing and generic types (#1486)
Browse files Browse the repository at this point in the history
  • Loading branch information
sbrunato authored Jan 17, 2025
1 parent 8e47f11 commit 7bfe7fb
Show file tree
Hide file tree
Showing 74 changed files with 812 additions and 917 deletions.
6 changes: 3 additions & 3 deletions docs/conf.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
import re
from datetime import datetime
from importlib.metadata import metadata
from typing import Any, Dict, List
from typing import Any

# -- General configuration ------------------------------------------------

Expand Down Expand Up @@ -168,7 +168,7 @@
"custom.css",
]

html_js_files: List[Any] = []
html_js_files: list[Any] = []

# Custom sidebar templates, must be a dictionary that maps document names
# to template names.
Expand All @@ -184,7 +184,7 @@

# -- Options for LaTeX output ---------------------------------------------

latex_elements: Dict[str, str] = {
latex_elements: dict[str, str] = {
# The paper size ('letterpaper' or 'a4paper').
#
# 'papersize': 'letterpaper',
Expand Down
86 changes: 43 additions & 43 deletions eodag/api/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
import shutil
import tempfile
from operator import itemgetter
from typing import TYPE_CHECKING, Any, Dict, Iterator, List, Optional, Tuple, Union
from typing import TYPE_CHECKING, Any, Iterator, Optional, Union

import geojson
import pkg_resources
Expand Down Expand Up @@ -185,7 +185,7 @@ def __init__(
self._plugins_manager.rebuild(self.providers_config)

# store pruned providers configs
self._pruned_providers_config: Dict[str, Any] = {}
self._pruned_providers_config: dict[str, Any] = {}
# filter out providers needing auth that have no credentials set
self._prune_providers_list()

Expand Down Expand Up @@ -335,7 +335,7 @@ def set_preferred_provider(self, provider: str) -> None:
new_priority = max_priority + 1
self._plugins_manager.set_priority(provider, new_priority)

def get_preferred_provider(self) -> Tuple[str, int]:
def get_preferred_provider(self) -> tuple[str, int]:
"""Get the provider currently set as the preferred one for searching
products, along with its priority.
Expand All @@ -351,7 +351,7 @@ def get_preferred_provider(self) -> Tuple[str, int]:
def update_providers_config(
self,
yaml_conf: Optional[str] = None,
dict_conf: Optional[Dict[str, Any]] = None,
dict_conf: Optional[dict[str, Any]] = None,
) -> None:
"""Update providers configuration with given input.
Can be used to add a provider to existing configuration or update
Expand Down Expand Up @@ -397,12 +397,12 @@ def add_provider(
name: str,
url: Optional[str] = None,
priority: Optional[int] = None,
search: Dict[str, Any] = {"type": "StacSearch"},
products: Dict[str, Any] = {
search: dict[str, Any] = {"type": "StacSearch"},
products: dict[str, Any] = {
GENERIC_PRODUCT_TYPE: {"productType": "{productType}"}
},
download: Dict[str, Any] = {"type": "HTTPDownload", "auth_error_code": 401},
**kwargs: Dict[str, Any],
download: dict[str, Any] = {"type": "HTTPDownload", "auth_error_code": 401},
**kwargs: dict[str, Any],
):
"""Adds a new provider.
Expand All @@ -421,7 +421,7 @@ def add_provider(
:param download: Download :class:`~eodag.config.PluginConfig` mapping
:param kwargs: Additional :class:`~eodag.config.ProviderConfig` mapping
"""
conf_dict: Dict[str, Any] = {
conf_dict: dict[str, Any] = {
name: {
"url": url,
"search": {"type": "StacSearch", **search},
Expand Down Expand Up @@ -565,7 +565,7 @@ def set_locations_conf(self, locations_conf_path: str) -> None:
main_locations_config = locations_config[main_key]

logger.info("Locations configuration loaded from %s" % locations_conf_path)
self.locations_config: List[Dict[str, Any]] = main_locations_config
self.locations_config: list[dict[str, Any]] = main_locations_config
else:
logger.info(
"Could not load locations configuration from %s" % locations_conf_path
Expand All @@ -574,7 +574,7 @@ def set_locations_conf(self, locations_conf_path: str) -> None:

def list_product_types(
self, provider: Optional[str] = None, fetch_providers: bool = True
) -> List[Dict[str, Any]]:
) -> list[dict[str, Any]]:
"""Lists supported product types.
:param provider: (optional) The name of a provider that must support the product
Expand All @@ -588,7 +588,7 @@ def list_product_types(
# First, update product types list if possible
self.fetch_product_types_list(provider=provider)

product_types: List[Dict[str, Any]] = []
product_types: list[dict[str, Any]] = []

providers_configs = (
list(self.providers_config.values())
Expand Down Expand Up @@ -644,7 +644,7 @@ def fetch_product_types_list(self, provider: Optional[str] = None) -> None:
providers_to_fetch = [provider]

# providers discovery confs that are fetchable
providers_discovery_configs_fetchable: Dict[str, Any] = {}
providers_discovery_configs_fetchable: dict[str, Any] = {}
# check if any provider has not already been fetched for product types
already_fetched = True
for provider_to_fetch in providers_to_fetch:
Expand Down Expand Up @@ -767,7 +767,7 @@ def fetch_product_types_list(self, provider: Optional[str] = None) -> None:

def discover_product_types(
self, provider: Optional[str] = None
) -> Optional[Dict[str, Any]]:
) -> Optional[dict[str, Any]]:
"""Fetch providers for product types
:param provider: The name of a provider or provider-group to fetch. Defaults to
Expand All @@ -787,7 +787,7 @@ def discover_product_types(
raise UnsupportedProvider(
f"The requested provider is not (yet) supported: {provider}"
)
ext_product_types_conf: Dict[str, Any] = {}
ext_product_types_conf: dict[str, Any] = {}
providers_to_fetch = [
p
for p in (
Expand All @@ -800,7 +800,7 @@ def discover_product_types(
else self.available_providers()
)
]
kwargs: Dict[str, Any] = {}
kwargs: dict[str, Any] = {}
for provider in providers_to_fetch:
if hasattr(self.providers_config[provider], "search"):
search_plugin_config = self.providers_config[provider].search
Expand Down Expand Up @@ -841,7 +841,7 @@ def discover_product_types(
return sort_dict(ext_product_types_conf)

def update_product_types_list(
self, ext_product_types_conf: Dict[str, Optional[Dict[str, Dict[str, Any]]]]
self, ext_product_types_conf: dict[str, Optional[dict[str, dict[str, Any]]]]
) -> None:
"""Update eodag product types list
Expand Down Expand Up @@ -869,7 +869,7 @@ def update_product_types_list(
provider,
)
continue
new_product_types: List[str] = []
new_product_types: list[str] = []
for (
new_product_type,
new_product_type_conf,
Expand Down Expand Up @@ -932,7 +932,7 @@ def update_product_types_list(

def available_providers(
self, product_type: Optional[str] = None, by_group: bool = False
) -> List[str]:
) -> list[str]:
"""Gives the sorted list of the available providers or groups
The providers or groups are sorted first by their priority level in descending order,
Expand All @@ -959,7 +959,7 @@ def available_providers(

# If by_group is True, keep only the highest priority for each group
if by_group:
group_priority: Dict[str, int] = {}
group_priority: dict[str, int] = {}
for name, priority in providers:
if name not in group_priority or priority > group_priority[name]:
group_priority[name] = priority
Expand Down Expand Up @@ -1026,7 +1026,7 @@ def guess_product_type(
missionStartDate: Optional[str] = None,
missionEndDate: Optional[str] = None,
**kwargs: Any,
) -> List[str]:
) -> list[str]:
"""
Find EODAG product type IDs that best match a set of search parameters.
Expand Down Expand Up @@ -1084,7 +1084,7 @@ def guess_product_type(
query = p.parse(text)
results = searcher.search(query, limit=None)

guesses: List[Dict[str, str]] = [dict(r) for r in results or []]
guesses: list[dict[str, str]] = [dict(r) for r in results or []]

# datetime filtering
if missionStartDate or missionEndDate:
Expand Down Expand Up @@ -1125,8 +1125,8 @@ def search(
raise_errors: bool = False,
start: Optional[str] = None,
end: Optional[str] = None,
geom: Optional[Union[str, Dict[str, float], BaseGeometry]] = None,
locations: Optional[Dict[str, str]] = None,
geom: Optional[Union[str, dict[str, float], BaseGeometry]] = None,
locations: Optional[dict[str, str]] = None,
provider: Optional[str] = None,
count: bool = False,
**kwargs: Any,
Expand Down Expand Up @@ -1205,7 +1205,7 @@ def search(
items_per_page=items_per_page,
)

errors: List[Tuple[str, Exception]] = []
errors: list[tuple[str, Exception]] = []
# Loop over available providers and return the first non-empty results
for i, search_plugin in enumerate(search_plugins):
search_plugin.clear()
Expand Down Expand Up @@ -1234,8 +1234,8 @@ def search_iter_page(
items_per_page: int = DEFAULT_ITEMS_PER_PAGE,
start: Optional[str] = None,
end: Optional[str] = None,
geom: Optional[Union[str, Dict[str, float], BaseGeometry]] = None,
locations: Optional[Dict[str, str]] = None,
geom: Optional[Union[str, dict[str, float], BaseGeometry]] = None,
locations: Optional[dict[str, str]] = None,
**kwargs: Any,
) -> Iterator[SearchResult]:
"""Iterate over the pages of a products search.
Expand Down Expand Up @@ -1411,8 +1411,8 @@ def search_all(
items_per_page: Optional[int] = None,
start: Optional[str] = None,
end: Optional[str] = None,
geom: Optional[Union[str, Dict[str, float], BaseGeometry]] = None,
locations: Optional[Dict[str, str]] = None,
geom: Optional[Union[str, dict[str, float], BaseGeometry]] = None,
locations: Optional[dict[str, str]] = None,
**kwargs: Any,
) -> SearchResult:
"""Search and return all the products matching the search criteria.
Expand Down Expand Up @@ -1633,7 +1633,7 @@ def _fetch_external_product_type(self, provider: str, product_type: str):
if not getattr(plugin.config, "discover_product_types", {}).get("fetch_url"):
return None

kwargs: Dict[str, Any] = {"productType": product_type}
kwargs: dict[str, Any] = {"productType": product_type}

# append auth if needed
if getattr(plugin.config, "need_auth", False):
Expand All @@ -1651,11 +1651,11 @@ def _prepare_search(
self,
start: Optional[str] = None,
end: Optional[str] = None,
geom: Optional[Union[str, Dict[str, float], BaseGeometry]] = None,
locations: Optional[Dict[str, str]] = None,
geom: Optional[Union[str, dict[str, float], BaseGeometry]] = None,
locations: Optional[dict[str, str]] = None,
provider: Optional[str] = None,
**kwargs: Any,
) -> Tuple[List[Union[Search, Api]], Dict[str, Any]]:
) -> tuple[list[Union[Search, Api]], dict[str, Any]]:
"""Internal method to prepare the search kwargs and get the search plugins.
Product query:
Expand Down Expand Up @@ -1763,7 +1763,7 @@ def _prepare_search(

preferred_provider = self.get_preferred_provider()[0]

search_plugins: List[Union[Search, Api]] = []
search_plugins: list[Union[Search, Api]] = []
for plugin in self._plugins_manager.get_search_plugins(
product_type=product_type, provider=provider
):
Expand Down Expand Up @@ -1833,10 +1833,10 @@ def _do_search(
max_items_per_page,
)

results: List[EOProduct] = []
results: list[EOProduct] = []
total_results: Optional[int] = 0 if count else None

errors: List[Tuple[str, Exception]] = []
errors: list[tuple[str, Exception]] = []

try:
prep = PreparedSearch(count=count)
Expand Down Expand Up @@ -1984,7 +1984,7 @@ def crunch(self, results: SearchResult, **kwargs: Any) -> SearchResult:
return results

@staticmethod
def group_by_extent(searches: List[SearchResult]) -> List[SearchResult]:
def group_by_extent(searches: list[SearchResult]) -> list[SearchResult]:
"""Combines multiple SearchResults and return a list of SearchResults grouped
by extent (i.e. bounding box).
Expand All @@ -1993,7 +1993,7 @@ def group_by_extent(searches: List[SearchResult]) -> List[SearchResult]:
"""
# Dict with extents as keys, each extent being defined by a str
# "{minx}{miny}{maxx}{maxy}" (each float rounded to 2 dec).
products_grouped_by_extent: Dict[str, Any] = {}
products_grouped_by_extent: dict[str, Any] = {}

for search in searches:
for product in search:
Expand All @@ -2015,7 +2015,7 @@ def download_all(
wait: float = DEFAULT_DOWNLOAD_WAIT,
timeout: float = DEFAULT_DOWNLOAD_TIMEOUT,
**kwargs: Unpack[DownloadConf],
) -> List[str]:
) -> list[str]:
"""Download all products resulting from a search.
:param search_result: A collection of EO products resulting from a search
Expand Down Expand Up @@ -2273,7 +2273,7 @@ def list_queryables(
properties, associating parameters to their annotated type, and a additional_properties attribute
"""
# only fetch providers if product type is not found
available_product_types: List[str] = [
available_product_types: list[str] = [
pt["ID"]
for pt in self.list_product_types(provider=provider, fetch_providers=False)
]
Expand Down Expand Up @@ -2310,7 +2310,7 @@ def list_queryables(

for plugin in self._plugins_manager.get_search_plugins(product_type, provider):
# attach product type config
product_type_configs: Dict[str, Any] = {}
product_type_configs: dict[str, Any] = {}
if product_type:
self._attach_product_type_config(plugin, product_type)
product_type_configs[product_type] = plugin.config.product_type_config
Expand Down Expand Up @@ -2354,15 +2354,15 @@ def list_queryables(
**queryable_properties,
)

def available_sortables(self) -> Dict[str, Optional[ProviderSortables]]:
def available_sortables(self) -> dict[str, Optional[ProviderSortables]]:
"""For each provider, gives its available sortable parameter(s) and its maximum
number of them if it supports the sorting feature, otherwise gives None.
:returns: A dictionary with providers as keys and dictionary of sortable parameter(s) and
its (their) maximum number as value(s).
:raises: :class:`~eodag.utils.exceptions.UnsupportedProvider`
"""
sortables: Dict[str, Optional[ProviderSortables]] = {}
sortables: dict[str, Optional[ProviderSortables]] = {}
provider_search_plugins = self._plugins_manager.get_search_plugins()
for provider_search_plugin in provider_search_plugins:
provider = provider_search_plugin.provider
Expand Down
10 changes: 5 additions & 5 deletions eodag/api/product/_assets.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@

import re
from collections import UserDict
from typing import TYPE_CHECKING, Any, Dict, List, Optional
from typing import TYPE_CHECKING, Any, Optional

from eodag.utils.exceptions import NotAvailableError
from eodag.utils.repr import dict_to_html_table
Expand All @@ -45,18 +45,18 @@ def __init__(self, product: EOProduct, *args: Any, **kwargs: Any) -> None:
self.product = product
super(AssetsDict, self).__init__(*args, **kwargs)

def __setitem__(self, key: str, value: Dict[str, Any]) -> None:
def __setitem__(self, key: str, value: dict[str, Any]) -> None:
super().__setitem__(key, Asset(self.product, key, value))

def as_dict(self) -> Dict[str, Any]:
def as_dict(self) -> dict[str, Any]:
"""Builds a representation of AssetsDict to enable its serialization
:returns: The representation of a :class:`~eodag.api.product._assets.AssetsDict`
as a Python dict
"""
return {k: v.as_dict() for k, v in self.data.items()}

def get_values(self, asset_filter: str = "") -> List[Asset]:
def get_values(self, asset_filter: str = "") -> list[Asset]:
"""
retrieves the assets matching the given filter
Expand Down Expand Up @@ -138,7 +138,7 @@ def __init__(self, product: EOProduct, key: str, *args: Any, **kwargs: Any) -> N
self.key = key
super(Asset, self).__init__(*args, **kwargs)

def as_dict(self) -> Dict[str, Any]:
def as_dict(self) -> dict[str, Any]:
"""Builds a representation of Asset to enable its serialization
:returns: The representation of a :class:`~eodag.api.product._assets.Asset` as a
Expand Down
Loading

0 comments on commit 7bfe7fb

Please sign in to comment.