2
2
3
3
from collections import ChainMap
4
4
from copy import copy
5
- from typing import TYPE_CHECKING , Any , Dict , List , Optional , Tuple , Union , cast
5
+ from typing import TYPE_CHECKING , Any , cast
6
6
7
7
import pystac
8
8
11
11
from pystac .stac_object import STACObject
12
12
13
13
14
- def get_cache_key (stac_object : STACObject ) -> Tuple [str , bool ]:
14
+ def get_cache_key (stac_object : STACObject ) -> tuple [str , bool ]:
15
15
"""Produce a cache key for the given STAC object.
16
16
17
17
If a self href is set, use that as the cache key.
@@ -27,8 +27,8 @@ def get_cache_key(stac_object: STACObject) -> Tuple[str, bool]:
27
27
if href is not None :
28
28
return href , True
29
29
else :
30
- ids : List [str ] = []
31
- obj : Optional [ pystac .STACObject ] = stac_object
30
+ ids : list [str ] = []
31
+ obj : pystac .STACObject | None = stac_object
32
32
while obj is not None :
33
33
ids .append (obj .id )
34
34
obj = obj .get_parent ()
@@ -61,23 +61,23 @@ class ResolvedObjectCache:
61
61
to collections.
62
62
"""
63
63
64
- id_keys_to_objects : Dict [str , STACObject ]
64
+ id_keys_to_objects : dict [str , STACObject ]
65
65
"""Existing cache of a key made up of the STACObject and it's parents IDs mapped
66
66
to the cached STACObject."""
67
67
68
- hrefs_to_objects : Dict [str , STACObject ]
68
+ hrefs_to_objects : dict [str , STACObject ]
69
69
"""STAC Object HREFs matched to their cached object."""
70
70
71
- ids_to_collections : Dict [str , Collection ]
71
+ ids_to_collections : dict [str , Collection ]
72
72
"""Map of collection IDs to collections."""
73
73
74
- _collection_cache : Optional [ " ResolvedObjectCollectionCache" ]
74
+ _collection_cache : ResolvedObjectCollectionCache | None
75
75
76
76
def __init__ (
77
77
self ,
78
- id_keys_to_objects : Optional [ Dict [ str , STACObject ]] = None ,
79
- hrefs_to_objects : Optional [ Dict [ str , STACObject ]] = None ,
80
- ids_to_collections : Optional [ Dict [ str , Collection ]] = None ,
78
+ id_keys_to_objects : dict [ str , STACObject ] | None = None ,
79
+ hrefs_to_objects : dict [ str , STACObject ] | None = None ,
80
+ ids_to_collections : dict [ str , Collection ] | None = None ,
81
81
):
82
82
self .id_keys_to_objects = id_keys_to_objects or {}
83
83
self .hrefs_to_objects = hrefs_to_objects or {}
@@ -111,7 +111,7 @@ def get_or_cache(self, obj: STACObject) -> STACObject:
111
111
self .cache (obj )
112
112
return obj
113
113
114
- def get (self , obj : STACObject ) -> Optional [ STACObject ] :
114
+ def get (self , obj : STACObject ) -> STACObject | None :
115
115
"""Get the cached object that has the same cache key as the given object.
116
116
117
117
Args:
@@ -128,7 +128,7 @@ def get(self, obj: STACObject) -> Optional[STACObject]:
128
128
else :
129
129
return self .id_keys_to_objects .get (key )
130
130
131
- def get_by_href (self , href : str ) -> Optional [ STACObject ] :
131
+ def get_by_href (self , href : str ) -> STACObject | None :
132
132
"""Gets the cached object at href.
133
133
134
134
Args:
@@ -139,7 +139,7 @@ def get_by_href(self, href: str) -> Optional[STACObject]:
139
139
"""
140
140
return self .hrefs_to_objects .get (href )
141
141
142
- def get_collection_by_id (self , id : str ) -> Optional [ Collection ] :
142
+ def get_collection_by_id (self , id : str ) -> Collection | None :
143
143
"""Retrieved a cached Collection by its ID.
144
144
145
145
Args:
@@ -199,7 +199,7 @@ def as_collection_cache(self) -> CollectionCache:
199
199
200
200
@staticmethod
201
201
def merge (
202
- first : " ResolvedObjectCache" , second : " ResolvedObjectCache"
202
+ first : ResolvedObjectCache , second : ResolvedObjectCache
203
203
) -> ResolvedObjectCache :
204
204
"""Merges two ResolvedObjectCache.
205
205
@@ -247,32 +247,30 @@ class CollectionCache:
247
247
in common properties.
248
248
"""
249
249
250
- cached_ids : Dict [str , Union [ Collection , Dict [str , Any ] ]]
251
- cached_hrefs : Dict [str , Union [ Collection , Dict [str , Any ] ]]
250
+ cached_ids : dict [str , Collection | dict [str , Any ]]
251
+ cached_hrefs : dict [str , Collection | dict [str , Any ]]
252
252
253
253
def __init__ (
254
254
self ,
255
- cached_ids : Optional [ Dict [ str , Union [ Collection , Dict [str , Any ]]]] = None ,
256
- cached_hrefs : Optional [ Dict [ str , Union [ Collection , Dict [str , Any ]]]] = None ,
255
+ cached_ids : dict [ str , Collection | dict [str , Any ]] | None = None ,
256
+ cached_hrefs : dict [ str , Collection | dict [str , Any ]] | None = None ,
257
257
):
258
258
self .cached_ids = cached_ids or {}
259
259
self .cached_hrefs = cached_hrefs or {}
260
260
261
- def get_by_id (
262
- self , collection_id : str
263
- ) -> Optional [Union [Collection , Dict [str , Any ]]]:
261
+ def get_by_id (self , collection_id : str ) -> Collection | dict [str , Any ] | None :
264
262
return self .cached_ids .get (collection_id )
265
263
266
- def get_by_href (self , href : str ) -> Optional [ Union [ Collection , Dict [str , Any ]]] :
264
+ def get_by_href (self , href : str ) -> Collection | dict [str , Any ] | None :
267
265
return self .cached_hrefs .get (href )
268
266
269
267
def contains_id (self , collection_id : str ) -> bool :
270
268
return collection_id in self .cached_ids
271
269
272
270
def cache (
273
271
self ,
274
- collection : Union [ Collection , Dict [str , Any ] ],
275
- href : Optional [ str ] = None ,
272
+ collection : Collection | dict [str , Any ],
273
+ href : str | None = None ,
276
274
) -> None :
277
275
"""Caches a collection JSON."""
278
276
if isinstance (collection , pystac .Collection ):
@@ -290,22 +288,20 @@ class ResolvedObjectCollectionCache(CollectionCache):
290
288
def __init__ (
291
289
self ,
292
290
resolved_object_cache : ResolvedObjectCache ,
293
- cached_ids : Optional [ Dict [ str , Union [ Collection , Dict [str , Any ]]]] = None ,
294
- cached_hrefs : Optional [ Dict [ str , Union [ Collection , Dict [str , Any ]]]] = None ,
291
+ cached_ids : dict [ str , Collection | dict [str , Any ]] | None = None ,
292
+ cached_hrefs : dict [ str , Collection | dict [str , Any ]] | None = None ,
295
293
):
296
294
super ().__init__ (cached_ids , cached_hrefs )
297
295
self .resolved_object_cache = resolved_object_cache
298
296
299
- def get_by_id (
300
- self , collection_id : str
301
- ) -> Optional [Union [Collection , Dict [str , Any ]]]:
297
+ def get_by_id (self , collection_id : str ) -> Collection | dict [str , Any ] | None :
302
298
result = self .resolved_object_cache .get_collection_by_id (collection_id )
303
299
if result is None :
304
300
return super ().get_by_id (collection_id )
305
301
else :
306
302
return result
307
303
308
- def get_by_href (self , href : str ) -> Optional [ Union [ Collection , Dict [str , Any ]]] :
304
+ def get_by_href (self , href : str ) -> Collection | dict [str , Any ] | None :
309
305
result = self .resolved_object_cache .get_by_href (href )
310
306
if result is None :
311
307
return super ().get_by_href (href )
@@ -319,16 +315,16 @@ def contains_id(self, collection_id: str) -> bool:
319
315
320
316
def cache (
321
317
self ,
322
- collection : Union [ Collection , Dict [str , Any ] ],
323
- href : Optional [ str ] = None ,
318
+ collection : Collection | dict [str , Any ],
319
+ href : str | None = None ,
324
320
) -> None :
325
321
super ().cache (collection , href )
326
322
327
323
@staticmethod
328
324
def merge (
329
325
resolved_object_cache : ResolvedObjectCache ,
330
- first : Optional [ " ResolvedObjectCollectionCache" ] ,
331
- second : Optional [ " ResolvedObjectCollectionCache" ] ,
326
+ first : ResolvedObjectCollectionCache | None ,
327
+ second : ResolvedObjectCollectionCache | None ,
332
328
) -> ResolvedObjectCollectionCache :
333
329
first_cached_ids = {}
334
330
if first is not None :
0 commit comments