Skip to content

Commit

Permalink
Fix type errors
Browse files Browse the repository at this point in the history
  • Loading branch information
RenataMuellerC2C committed Dec 11, 2024
1 parent e0ef724 commit b07efb1
Show file tree
Hide file tree
Showing 2 changed files with 34 additions and 23 deletions.
27 changes: 15 additions & 12 deletions geoportal/c2cgeoportal_geoportal/views/profile.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,14 +28,15 @@
import json
import logging
import math
import urllib.parse
from decimal import Decimal
from json.decoder import JSONDecodeError
from typing import Any

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

Expand Down Expand Up @@ -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)

Expand All @@ -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:
Expand Down
30 changes: 19 additions & 11 deletions geoportal/c2cgeoportal_geoportal/views/raster.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,14 +32,15 @@
import math
import os
import traceback
import urllib.parse
from json.decoder import JSONDecodeError
from typing import TYPE_CHECKING, Any

import numpy
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

Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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)

Expand Down

0 comments on commit b07efb1

Please sign in to comment.