33This module contains functions for transforming geospatial coordinates,
44such as converting bounding boxes to polygon representations.
55"""
6- from typing import List
6+ from typing import Any , Dict , List , Optional , Set , Union
7+
8+ from stac_fastapi .types .stac import Item
79
810MAX_LIMIT = 10000
911
@@ -21,3 +23,113 @@ def bbox2polygon(b0: float, b1: float, b2: float, b3: float) -> List[List[List[f
2123 List[List[List[float]]]: A polygon represented as a list of lists of coordinates.
2224 """
2325 return [[[b0 , b1 ], [b2 , b1 ], [b2 , b3 ], [b0 , b3 ], [b0 , b1 ]]]
26+
27+
28+ # copied from stac-fastapi-pgstac
29+ # https://github.com/stac-utils/stac-fastapi-pgstac/blob/26f6d918eb933a90833f30e69e21ba3b4e8a7151/stac_fastapi/pgstac/utils.py#L10-L116
30+ def filter_fields ( # noqa: C901
31+ item : Union [Item , Dict [str , Any ]],
32+ include : Optional [Set [str ]] = None ,
33+ exclude : Optional [Set [str ]] = None ,
34+ ) -> Item :
35+ """Preserve and remove fields as indicated by the fields extension include/exclude sets.
36+
37+ Returns a shallow copy of the Item with the fields filtered.
38+
39+ This will not perform a deep copy; values of the original item will be referenced
40+ in the return item.
41+ """
42+ if not include and not exclude :
43+ return item
44+
45+ # Build a shallow copy of included fields on an item, or a sub-tree of an item
46+ def include_fields (
47+ source : Dict [str , Any ], fields : Optional [Set [str ]]
48+ ) -> Dict [str , Any ]:
49+ if not fields :
50+ return source
51+
52+ clean_item : Dict [str , Any ] = {}
53+ for key_path in fields or []:
54+ key_path_parts = key_path .split ("." )
55+ key_root = key_path_parts [0 ]
56+ if key_root in source :
57+ if isinstance (source [key_root ], dict ) and len (key_path_parts ) > 1 :
58+ # The root of this key path on the item is a dict, and the
59+ # key path indicates a sub-key to be included. Walk the dict
60+ # from the root key and get the full nested value to include.
61+ value = include_fields (
62+ source [key_root ], fields = {"." .join (key_path_parts [1 :])}
63+ )
64+
65+ if isinstance (clean_item .get (key_root ), dict ):
66+ # A previously specified key and sub-keys may have been included
67+ # already, so do a deep merge update if the root key already exists.
68+ dict_deep_update (clean_item [key_root ], value )
69+ else :
70+ # The root key does not exist, so add it. Fields
71+ # extension only allows nested referencing on dicts, so
72+ # this won't overwrite anything.
73+ clean_item [key_root ] = value
74+ else :
75+ # The item value to include is not a dict, or, it is a dict but the
76+ # key path is for the whole value, not a sub-key. Include the entire
77+ # value in the cleaned item.
78+ clean_item [key_root ] = source [key_root ]
79+ else :
80+ # The key, or root key of a multi-part key, is not present in the item,
81+ # so it is ignored
82+ pass
83+ return clean_item
84+
85+ # For an item built up for included fields, remove excluded fields. This
86+ # modifies `source` in place.
87+ def exclude_fields (source : Dict [str , Any ], fields : Optional [Set [str ]]) -> None :
88+ for key_path in fields or []:
89+ key_path_part = key_path .split ("." )
90+ key_root = key_path_part [0 ]
91+ if key_root in source :
92+ if isinstance (source [key_root ], dict ) and len (key_path_part ) > 1 :
93+ # Walk the nested path of this key to remove the leaf-key
94+ exclude_fields (
95+ source [key_root ], fields = {"." .join (key_path_part [1 :])}
96+ )
97+ # If, after removing the leaf-key, the root is now an empty
98+ # dict, remove it entirely
99+ if not source [key_root ]:
100+ del source [key_root ]
101+ else :
102+ # The key's value is not a dict, or there is no sub-key to remove. The
103+ # entire key can be removed from the source.
104+ source .pop (key_root , None )
105+
106+ # Coalesce incoming type to a dict
107+ item = dict (item )
108+
109+ clean_item = include_fields (item , include )
110+
111+ # If, after including all the specified fields, there are no included properties,
112+ # return just id and collection.
113+ if not clean_item :
114+ return Item ({"id" : item ["id" ], "collection" : item ["collection" ]})
115+
116+ exclude_fields (clean_item , exclude )
117+
118+ return Item (** clean_item )
119+
120+
121+ def dict_deep_update (merge_to : Dict [str , Any ], merge_from : Dict [str , Any ]) -> None :
122+ """Perform a deep update of two dicts.
123+
124+ merge_to is updated in-place with the values from merge_from.
125+ merge_from values take precedence over existing values in merge_to.
126+ """
127+ for k , v in merge_from .items ():
128+ if (
129+ k in merge_to
130+ and isinstance (merge_to [k ], dict )
131+ and isinstance (merge_from [k ], dict )
132+ ):
133+ dict_deep_update (merge_to [k ], merge_from [k ])
134+ else :
135+ merge_to [k ] = v
0 commit comments