Patch/update rio tiler req (#247)

* fix type hints and update rio-tiler requirements

* update changelog

Author: vincentsarago

CHANGES.md

@@ -1,5 +1,10 @@
 ## Unreleased
 
+## 9.1.0 (2026-02-02)
+
+* update: rio-tiler requirements 
+* update: type hints
+
 ## 9.0.2 (2025-12-15)
 
 * add support for python 3.14

cogeo_mosaic/backends/base.py

@@ -3,8 +3,9 @@
 import abc
 import itertools
 import warnings
+from collections.abc import Sequence
 from threading import Lock
-from typing import Any, Dict, List, Optional, Sequence, Type, Union
+from typing import Any
 
 import attr
 from cachetools import TTLCache, cached
@@ -25,7 +26,7 @@
 from cogeo_mosaic.utils import bbox_union
 
 
-def _convert_to_mosaicjson(value: Union[Dict, MosaicJSON]):
+def _convert_to_mosaicjson(value: dict | MosaicJSON):
     if value is not None:
         return MosaicJSON(**dict(value))
 
@@ -55,18 +56,16 @@ class MosaicJSONBackend(BaseBackend):
     minzoom: int = attr.ib(default=None)
     maxzoom: int = attr.ib(default=None)
 
-    reader: Union[
-        Type[BaseReader],
-        Type[MultiBaseReader],
-        Type[MultiBandReader],
-    ] = attr.ib(default=Reader)
-    reader_options: Dict = attr.ib(factory=dict)
+    reader: type[BaseReader] | type[MultiBaseReader] | type[MultiBandReader] = attr.ib(
+        default=Reader
+    )
+    reader_options: dict = attr.ib(factory=dict)
 
     bounds: BBox = attr.ib(init=False)
     crs: CRS = attr.ib(init=False)
 
     _backend_name: str
-    _file_byte_size: Optional[int] = 0
+    _file_byte_size: int | None = 0
 
     def __attrs_post_init__(self):
         """Post Init: if not passed in init, try to read from self.input."""
@@ -103,7 +102,7 @@ def write(self, overwrite: bool = True):
 
     def update(
         self,
-        features: Sequence[Dict],
+        features: Sequence[dict],
         add_first: bool = True,
         quiet: bool = False,
         **kwargs,
@@ -146,7 +145,7 @@ def update(
         self.bounds = bounds
         self.write(overwrite=True)
 
-    def assets_for_tile(self, x: int, y: int, z: int, **kwargs: Any) -> List[str]:
+    def assets_for_tile(self, x: int, y: int, z: int, **kwargs: Any) -> list[str]:
         """Retrieve assets for tile."""
         mosaic_tms = self.mosaic_def.tilematrixset or WEB_MERCATOR_TMS
         if self.tms == mosaic_tms:
@@ -168,9 +167,9 @@ def assets_for_point(
         self,
         lng: float,
         lat: float,
-        coord_crs: Optional[CRS] = None,
+        coord_crs: CRS | None = None,
         **kwargs: Any,
-    ) -> List[str]:
+    ) -> list[str]:
         """Retrieve assets for point."""
         mosaic_tms = self.mosaic_def.tilematrixset or WEB_MERCATOR_TMS
 
@@ -196,9 +195,9 @@ def assets_for_bbox(
         ymin: float,
         xmax: float,
         ymax: float,
-        coord_crs: Optional[CRS] = None,
+        coord_crs: CRS | None = None,
         **kwargs,
-    ) -> List[str]:
+    ) -> list[str]:
         """Retrieve assets for bbox."""
         mosaic_tms = self.mosaic_def.tilematrixset or WEB_MERCATOR_TMS
 
@@ -241,7 +240,7 @@ def assets_for_bbox(
         ),
         lock=Lock(),
     )
-    def get_assets(self, x: int, y: int, z: int, reverse: bool = False) -> List[str]:
+    def get_assets(self, x: int, y: int, z: int, reverse: bool = False) -> list[str]:
         """Find assets."""
         quadkeys = self.find_quadkeys(Tile(x=x, y=y, z=z), self.quadkey_zoom)
         assets = list(
@@ -259,7 +258,7 @@ def get_assets(self, x: int, y: int, z: int, reverse: bool = False) -> List[str]
 
         return assets
 
-    def find_quadkeys(self, tile: Tile, quadkey_zoom: int) -> List[str]:
+    def find_quadkeys(self, tile: Tile, quadkey_zoom: int) -> list[str]:
         """
         Find quadkeys at desired zoom for tile
 
@@ -273,7 +272,7 @@ def find_quadkeys(self, tile: Tile, quadkey_zoom: int) -> List[str]:
         Returns
         -------
         list
-            List[str] of quadkeys
+            list[str] of quadkeys
 
         """
         mosaic_tms = self.mosaic_def.tilematrixset or WEB_MERCATOR_TMS
@@ -318,7 +317,7 @@ def mosaicid(self) -> str:
         return get_hash(**self.mosaic_def.model_dump(exclude_none=True))
 
     @property
-    def _quadkeys(self) -> List[str]:
+    def _quadkeys(self) -> list[str]:
         """Return the list of quadkey tiles."""
         return list(self.mosaic_def.tiles)
 

cogeo_mosaic/backends/dynamodb.py

@@ -5,9 +5,10 @@
 import os
 import re
 import warnings
+from collections.abc import Sequence
 from decimal import Decimal
 from threading import Lock
-from typing import Any, Dict, List, Sequence
+from typing import Any
 from urllib.parse import urlparse
 
 import attr
@@ -136,7 +137,7 @@ def write(self, overwrite: bool = False, **kwargs: Any):
                 )
             self.delete()
 
-        items: List[Dict[str, Any]] = []
+        items: list[dict[str, Any]] = []
 
         # Create Metadata item
         # Note: `parse_float=Decimal` is required because DynamoDB requires all numbers to be
@@ -158,7 +159,7 @@ def write(self, overwrite: bool = False, **kwargs: Any):
 
     def update(
         self,
-        features: Sequence[Dict],
+        features: Sequence[dict],
         add_first: bool = True,
         quiet: bool = False,
         **kwargs,
@@ -193,7 +194,7 @@ def update(
         )
         self.bounds = bounds
 
-        items: List[Dict[str, Any]] = []
+        items: list[dict[str, Any]] = []
 
         # Create Metadata item
         # Note: `parse_float=Decimal` is required because DynamoDB requires all numbers to be
@@ -222,7 +223,7 @@ def update(
         key=lambda self, x, y, z: hashkey(self.input, x, y, z, self.mosaicid),
         lock=Lock(),
     )
-    def get_assets(self, x: int, y: int, z: int) -> List[str]:
+    def get_assets(self, x: int, y: int, z: int) -> list[str]:
         """Find assets."""
         quadkeys = self.find_quadkeys(Tile(x=x, y=y, z=z), self.quadkey_zoom)
         assets = list(
@@ -238,7 +239,7 @@ def get_assets(self, x: int, y: int, z: int) -> List[str]:
         return assets
 
     @property
-    def _quadkeys(self) -> List[str]:
+    def _quadkeys(self) -> list[str]:
         """Return the list of quadkey tiles."""
         resp = self.table.query(
             KeyConditionExpression=Key("mosaicId").eq(self.mosaic_name),
@@ -288,7 +289,7 @@ def _create_table(self, billing_mode: str = "PAY_PER_REQUEST", **kwargs: Any):
             warnings.warn("Unable to create table.")
             return
 
-    def _write_items(self, items: List[Dict]):
+    def _write_items(self, items: list[dict]):
         with self.table.batch_writer() as batch:
             with click.progressbar(
                 items,
@@ -299,7 +300,7 @@ def _write_items(self, items: List[Dict]):
                 for item in progitems:
                     batch.put_item(item)
 
-    def _fetch_dynamodb(self, quadkey: str) -> Dict:
+    def _fetch_dynamodb(self, quadkey: str) -> dict:
         try:
             return self.table.get_item(
                 Key={"mosaicId": self.mosaic_name, "quadkey": quadkey}

cogeo_mosaic/backends/sqlite.py

@@ -5,9 +5,9 @@
 import re
 import sqlite3
 import warnings
+from collections.abc import Sequence
 from pathlib import Path
 from threading import Lock
-from typing import Dict, List, Sequence
 from urllib.parse import urlparse
 
 import attr
@@ -215,7 +215,7 @@ def write(self, overwrite: bool = False):
 
     def update(
         self,
-        features: Sequence[Dict],
+        features: Sequence[dict],
         add_first: bool = True,
         quiet: bool = False,
         **kwargs,
@@ -315,7 +315,7 @@ def update(
         key=lambda self, x, y, z: hashkey(self.input, x, y, z, self.mosaicid),
         lock=Lock(),
     )
-    def get_assets(self, x: int, y: int, z: int) -> List[str]:
+    def get_assets(self, x: int, y: int, z: int) -> list[str]:
         """Find assets."""
         mercator_tile = morecantile.Tile(x=x, y=y, z=z)
         quadkeys = self.find_quadkeys(mercator_tile, self.quadkey_zoom)
@@ -330,23 +330,23 @@ def get_assets(self, x: int, y: int, z: int) -> List[str]:
         return assets
 
     @property
-    def _quadkeys(self) -> List[str]:
+    def _quadkeys(self) -> list[str]:
         """Return the list of quadkey tiles."""
         with self.db:
             rows = self.db.execute(
                 f'SELECT quadkey FROM "{self.mosaic_name}";',
             ).fetchall()
         return [r["quadkey"] for r in rows]
 
-    def _fetch_metadata(self) -> Dict:
+    def _fetch_metadata(self) -> dict:
         with self.db:
             row = self.db.execute(
                 f"SELECT * FROM {self._metadata_table} WHERE name=?;",
                 (self.mosaic_name,),
             ).fetchone()
             return dict(row) if row else {}
 
-    def _fetch(self, quadkey: str) -> List:
+    def _fetch(self, quadkey: str) -> list:
         with self.db:
             row = self.db.execute(
                 f'SELECT assets FROM "{self.mosaic_name}" WHERE quadkey=?;', (quadkey,)
@@ -377,7 +377,7 @@ def delete(self):
     def list_mosaics_in_db(
         cls,
         db_path: str,
-    ) -> List[str]:
+    ) -> list[str]:
         """List Mosaic tables in SQLite database.
 
         Args:

cogeo_mosaic/backends/stac.py

@@ -2,8 +2,8 @@
 
 import json
 import os
+from collections.abc import Sequence
 from threading import Lock
-from typing import Dict, List, Optional, Sequence, Type
 
 import attr
 import httpx
@@ -22,7 +22,7 @@
 from cogeo_mosaic.mosaic import MosaicJSON
 
 
-def default_stac_accessor(feature: Dict):
+def default_stac_accessor(feature: dict):
     """Return feature identifier."""
     link = list(filter(lambda link: link["rel"] == "self", feature["links"]))
     if link:
@@ -58,26 +58,26 @@ class STACBackend(MosaicJSONBackend):
     """
 
     input: str = attr.ib()
-    query: Dict = attr.ib()
+    query: dict = attr.ib()
 
     minzoom: int = attr.ib()
     maxzoom: int = attr.ib()
 
     tms: TileMatrixSet = attr.ib(default=WEB_MERCATOR_TMS)
 
-    reader: Type[STACReader] = attr.ib(default=STACReader)
-    reader_options: Dict = attr.ib(factory=dict)
+    reader: type[STACReader] = attr.ib(default=STACReader)
+    reader_options: dict = attr.ib(factory=dict)
 
     bounds: BBox = attr.ib(init=False, default=(-180, -90, 180, 90))
     crs: CRS = attr.ib(init=False, default=WGS84_CRS)
 
     # STAC API related options
     # max_items |  next_link_key | limit
-    stac_api_options: Dict = attr.ib(factory=dict)
+    stac_api_options: dict = attr.ib(factory=dict)
 
     # Mosaic Creation options
     # e.g `accessor`
-    mosaic_options: Dict = attr.ib(factory=dict)
+    mosaic_options: dict = attr.ib(factory=dict)
 
     # Because the STACBackend is a Read-Only backend, there is no need for
     # mosaic_def to be in the init method.
@@ -126,7 +126,7 @@ def write(self, overwrite: bool = True):
 
     def update(
         self,
-        features: Sequence[Dict],
+        features: Sequence[dict],
         add_first: bool = True,
         quiet: bool = False,
         **kwargs,
@@ -135,7 +135,7 @@ def update(
         raise NotImplementedError
 
 
-def query_from_link(link: Dict, query: Dict):
+def query_from_link(link: dict, query: dict):
     """Handle Next Link."""
     q = query.copy()
     if link["method"] != "POST":
@@ -156,13 +156,13 @@ def query_from_link(link: Dict, query: Dict):
 )
 def _fetch(  # noqa: C901
     stac_url: str,
-    query: Dict,
-    max_items: Optional[int] = None,
-    next_link_key: Optional[str] = None,
+    query: dict,
+    max_items: int | None = None,
+    next_link_key: str | None = None,
     limit: int = 500,
-) -> List[Dict]:
+) -> list[dict]:
     """Call STAC API."""
-    features: List[Dict] = []
+    features: list[dict] = []
     stac_query = query.copy()
 
     headers = {
@@ -174,7 +174,7 @@ def _fetch(  # noqa: C901
     if "limit" not in stac_query:
         stac_query.update({"limit": limit})
 
-    def _stac_search(url: str, q: Dict):
+    def _stac_search(url: str, q: dict):
         try:
             r = httpx.post(url, headers=headers, json=q)
             r.raise_for_status()

cogeo_mosaic/backends/web.py

@@ -5,8 +5,8 @@
 """
 
 import json
+from collections.abc import Sequence
 from threading import Lock
-from typing import Dict, Sequence
 
 import attr
 import httpx
@@ -64,7 +64,7 @@ def write(self, overwrite: bool = True):
 
     def update(
         self,
-        features: Sequence[Dict],
+        features: Sequence[dict],
         add_first: bool = True,
         quiet: bool = False,
         **kwargs,