|
1 | 1 | """titiler.core utilities.""" |
2 | 2 |
|
3 | 3 | import warnings |
4 | | -from typing import Any, Optional, Sequence, Tuple, Union |
| 4 | +from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, TypeVar, Union |
| 5 | +from urllib.parse import urlencode |
5 | 6 |
|
6 | 7 | import numpy |
| 8 | +from fastapi.datastructures import QueryParams |
| 9 | +from fastapi.dependencies.utils import get_dependant, request_params_to_args |
7 | 10 | from geojson_pydantic.geometries import MultiPolygon, Polygon |
8 | 11 | from rasterio.dtypes import dtype_ranges |
9 | 12 | from rio_tiler.colormap import apply_cmap |
@@ -131,3 +134,52 @@ def bounds_to_geometry(bounds: BBox) -> Union[Polygon, MultiPolygon]: |
131 | 134 | coordinates=[pl.coordinates, pr.coordinates], |
132 | 135 | ) |
133 | 136 | return Polygon.from_bounds(*bounds) |
| 137 | + |
| 138 | + |
| 139 | +T = TypeVar("T") |
| 140 | + |
| 141 | +ValidParams = Dict[str, Any] |
| 142 | +Errors = List[Any] |
| 143 | + |
| 144 | + |
| 145 | +def get_dependency_query_params( |
| 146 | + dependency: Callable, |
| 147 | + params: Dict, |
| 148 | +) -> Tuple[ValidParams, Errors]: |
| 149 | + """Check QueryParams for Query dependency. |
| 150 | +
|
| 151 | + 1. `get_dependant` is used to get the query-parameters required by the `callable` |
| 152 | + 2. we use `request_params_to_args` to construct arguments needed to call the `callable` |
| 153 | + 3. we call the `callable` and catch any errors |
| 154 | +
|
| 155 | + Important: We assume the `callable` in not a co-routine. |
| 156 | + """ |
| 157 | + dep = get_dependant(path="", call=dependency) |
| 158 | + return request_params_to_args( |
| 159 | + dep.query_params, QueryParams(urlencode(params, doseq=True)) |
| 160 | + ) |
| 161 | + |
| 162 | + |
| 163 | +def deserialize_query_params( |
| 164 | + dependency: Callable[..., T], params: Dict |
| 165 | +) -> Tuple[T, Errors]: |
| 166 | + """Deserialize QueryParams for given dependency. |
| 167 | +
|
| 168 | + Parse params as query params and deserialize with dependency. |
| 169 | +
|
| 170 | + Important: We assume the `callable` in not a co-routine. |
| 171 | + """ |
| 172 | + values, errors = get_dependency_query_params(dependency, params) |
| 173 | + return dependency(**values), errors |
| 174 | + |
| 175 | + |
| 176 | +def extract_query_params(dependencies, params) -> Tuple[ValidParams, Errors]: |
| 177 | + """Extract query params given list of dependencies.""" |
| 178 | + values = {} |
| 179 | + errors = [] |
| 180 | + for dep in dependencies: |
| 181 | + dep_values, dep_errors = deserialize_query_params(dep, params) |
| 182 | + if dep_values: |
| 183 | + values.update(dep_values) |
| 184 | + errors += dep_errors |
| 185 | + return values, errors |
0 commit comments