From b07efb16828a634e6df4a9b45a5cf51e7f3cbb01 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Renata=20M=C3=BCller?= Date: Wed, 11 Dec 2024 09:58:42 +0100 Subject: [PATCH] Fix type errors --- .../c2cgeoportal_geoportal/views/profile.py | 27 +++++++++-------- .../c2cgeoportal_geoportal/views/raster.py | 30 ++++++++++++------- 2 files changed, 34 insertions(+), 23 deletions(-) diff --git a/geoportal/c2cgeoportal_geoportal/views/profile.py b/geoportal/c2cgeoportal_geoportal/views/profile.py index 202b0428c5..879a4494f7 100644 --- a/geoportal/c2cgeoportal_geoportal/views/profile.py +++ b/geoportal/c2cgeoportal_geoportal/views/profile.py @@ -28,6 +28,7 @@ import json import logging import math +import urllib.parse from decimal import Decimal from json.decoder import JSONDecodeError from typing import Any @@ -35,7 +36,7 @@ import geojson import pyramid.request import requests -from pyramid.httpexceptions import HTTPBadRequest, HTTPNotFound +from pyramid.httpexceptions import HTTPBadRequest, HTTPInternalServerError, HTTPNotFound from pyramid.i18n import TranslationStringFactory from pyramid.view import view_config @@ -70,19 +71,21 @@ def _to_filtered(points: list[dict[str, Any]], layers: list[str]) -> list[dict[s def _get_profile_service_data( self, layers: list[str], geom: dict[str, Any], rasters: dict[str, Any], nb_points: int - ) -> dict[str, Any]: - request = ( - f"{rasters[layers[0]]['url']}/profile.json?geom={geom}&nbPoints={nb_points}&distinct_points=true" - ) + ) -> list[dict[str, Any]]: + request = f"{rasters[layers[0]]['url']}/profile.json?{urllib.parse.urlencode({'geom': geom, 'nbPoints': nb_points, 'distinct_points': 'true'})}" response = requests.get(request, timeout=10) + if not response.ok: + _LOG.error("profile request %s failed with status code %s", request, response.status_code) + raise HTTPInternalServerError( + f"Failed to fetch profile data from internal request: \ + {response.status_code} {response.reason}" + ) + try: points = json.loads(response.content) - except (TypeError, JSONDecodeError): - _LOG.warning("profile request %s failed", request) - self.request.response.status_code = response.status_code - self.request.response.text = response.text - self.request.response.content_type = "text/plain" - return self.request.response + except (TypeError, JSONDecodeError) as exc: + _LOG.exception("profile request %s failed", request) + raise HTTPInternalServerError("Failed to decode JSON response from internal request") from exc return self._to_filtered(points, layers) @@ -98,7 +101,7 @@ def _compute_points(self) -> tuple[list[str], list[dict[str, Any]]]: geom = geojson.loads(self.request.params["geom"], object_hook=geojson.GeoJSON.to_instance) nb_points = int(self.request.params["nbPoints"]) coords = [] - service_results = [] + service_results: list[dict[str, Any]] = [] layers: list[str] if "layers" in self.request.params: diff --git a/geoportal/c2cgeoportal_geoportal/views/raster.py b/geoportal/c2cgeoportal_geoportal/views/raster.py index b3a55e9d47..bfa2e8c3f7 100644 --- a/geoportal/c2cgeoportal_geoportal/views/raster.py +++ b/geoportal/c2cgeoportal_geoportal/views/raster.py @@ -32,6 +32,7 @@ import math import os import traceback +import urllib.parse from json.decoder import JSONDecodeError from typing import TYPE_CHECKING, Any @@ -39,7 +40,7 @@ import pyramid.request import requests import zope.event.classhandler -from pyramid.httpexceptions import HTTPBadRequest, HTTPNotFound +from pyramid.httpexceptions import HTTPBadRequest, HTTPInternalServerError, HTTPNotFound from pyramid.view import view_config from rasterio.io import DatasetReader @@ -101,13 +102,14 @@ def raster(self) -> dict[str, Any]: layers = list(rasters.keys()) layers.sort() - result = {} + result: dict[str, Any] = {} service_layers = [layer for layer in layers if rasters[layer].get("type") == "external_url"] if len(service_layers) > 0: for layer in service_layers: - result.update(self._get_service_data(layer, lat, lon, rasters)) + service_result: dict[str, Any] = self._get_service_data(layer, lat, lon, rasters) + result.update(service_result) for ref in list(rasters.keys()): if ref not in service_layers: @@ -196,18 +198,24 @@ def get_index(index_: int) -> tuple[int, int]: def _get_service_data( self, layer: str, lat: float, lon: float, rasters: dict[str, Any] - ) -> decimal.Decimal | None: - request = f"{rasters[layer]['url']}/height?easting={lon}&northing={lat}" + ) -> dict[str, Any]: + request = ( + f"{rasters[layer]['url']}/height?{urllib.parse.urlencode({'easting': lon, 'northing': lat})}" + ) _LOG.info("Doing height request to %s", request) response = requests.get(request, timeout=10) + if not response.ok: + _LOG.error("Elevation request %s failed with status code %s", request, response.status_code) + raise HTTPInternalServerError( + f"Failed to fetch elevation data from the internal request: \ + {response.status_code} {response.reason}" + ) + try: result = json.loads(response.content).get(rasters[layer]["elevation_name"]) - except (TypeError, JSONDecodeError): - _LOG.warning("Height request to %s failed", request) - self.request.response.status_code = response.status_code - self.request.response.text = response.text - self.request.response.content_type = "text/plain" - return self.request.response + except (TypeError, JSONDecodeError) as exc: + _LOG.exception("Height request to %s failed", request) + raise HTTPInternalServerError("Failed to decode JSON response from the internal request") from exc set_common_headers(self.request, "raster", Cache.PUBLIC_NO)