Skip to content

Fix SonarQube Findings #9

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jdw-creare merged 1 commit into from
Jun 6, 2025
Merged

Fix SonarQube Findings #9

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions example/app.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,13 +14,13 @@
import numpy as np

# create some podpac nodes
data = np.random.rand(11, 21)
data = np.random.default_rng(1).random((11, 21))
lat = np.linspace(90, -90, 11)
lon = np.linspace(-180, 180, 21)
coords = podpac.Coordinates([lat, lon], dims=["lat", "lon"])
node1 = podpac.data.Array(source=data, coordinates=coords)

data2 = np.random.rand(11, 21)
data2 = np.random.default_rng(1).random((11, 21))
node2 = podpac.data.Array(source=data2, coordinates=coords)

# use podpac nodes to create some OGC layers
Expand Down Expand Up @@ -75,10 +75,12 @@ def api_home(endpoint):

app = servers.FlaskServer(__name__, ogcs=[NonFouoOGC, FouoOGC], home_func=api_home)


# add in some other endpoints.
@app.route("/")
def home():
return f'This is an example OGC flask app. See <a href="/ogc_full"> FULL </a> and <a href="/ogc"> PARTIAL </a> endpoints.'
return """This is an example OGC flask app.
See <a href="/ogc_full"> FULL </a> and <a href="/ogc"> PARTIAL </a> endpoints."""


@app.route("/layers/<layer>")
Expand Down
295 changes: 152 additions & 143 deletions ogc/core.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

Currently holds some definitions for interface classes.
"""

import gc
import logging
import traitlets as tl
Expand All @@ -21,16 +22,15 @@
LOAD_FAILURE = "Failed to load and validate: "
INVALID_ARGUMENTS = "Invalid arguments"


class OGC(tl.HasTraits):

wms_capabilities = tl.Instance(klass=wms_response_1_3_0.Capabilities)
wcs_capabilities = tl.Instance(klass=wcs_response_1_0_0.Capabilities)

endpoint = tl.Unicode(default_value="/ogc", allow_none=True)
service_title = tl.Unicode(default_value="OGC Server", allow_none=True)
service_abstract = tl.Unicode(
default_value="An example OGC Server", allow_none=True
)
service_abstract = tl.Unicode(default_value="An example OGC Server", allow_none=True)
server_address = tl.Unicode(default_value="http://127.0.0.1:5000", allow_none=True)
service_group_title = tl.Unicode(default_value="Data Products", allow_none=True)

Expand Down Expand Up @@ -73,22 +73,83 @@ def get_coverage_from_id(self, identifier):
exception_text="Invalid coverage {}".format(identifier),
)

def handle_wcs_kv(self, args):
if args["request"] == "GetCapabilities":
get_capabilities = wcs_request_1_0_0.GetCapabilities()
try:
get_capabilities.load_from_kv(args)
get_capabilities.validate()
except Exception:
logger.error(LOAD_FAILURE, exc_info=True)
raise WCSException(exception_text=INVALID_ARGUMENTS)
def get_capabilities_wcs(self, args):
get_capabilities = wcs_request_1_0_0.GetCapabilities()
try:
get_capabilities.load_from_kv(args)
get_capabilities.validate()
except Exception:
logger.error(LOAD_FAILURE, exc_info=True)
raise WCSException(exception_text=INVALID_ARGUMENTS)

capabilities = self.wcs_capabilities

if args["base_url"]:
capabilities.base_url = args["base_url"]

return capabilities.to_xml()

def describe_coverage_wcs(self, args, wcs_request, wcs_response):
describe_coverage = wcs_request.DescribeCoverage()
try:
describe_coverage.load_from_kv(args)
describe_coverage.validate()
except Exception:
logger.error(LOAD_FAILURE, exc_info=True)
raise WCSException(exception_text=INVALID_ARGUMENTS)

coverages = [self.get_coverage_from_id(identifier.value) for identifier in describe_coverage.identifiers]
coverage_description = wcs_response.CoverageDescription(coverages=coverages)

return coverage_description.to_xml()

def get_coverage_wcs(self, args, wcs_request):
get_coverage = wcs_request.GetCoverage()
try:
get_coverage.load_from_kv(args)
get_coverage.validate()
except Exception:
logger.error(LOAD_FAILURE, exc_info=True)
raise WCSException(exception_text=INVALID_ARGUMENTS)

coverage = self.get_coverage_from_id(get_coverage.identifier.value)

from dateutil.parser import parse

if get_coverage.width == 0:
raise WCSException(
exception_code="InvalidParameterValue",
locator="VERSION",
exception_text="Grid coordinates x_size must be greater than 0",
)
if get_coverage.height == 0:
raise WCSException(
exception_code="InvalidParameterValue",
locator="VERSION",
exception_text="Grid coordinates y_size must be greater than 0",
)
if get_coverage.height * get_coverage.width > settings.MAX_GRID_COORDS_REQUEST_SIZE:
raise WCSException(
exception_code="InvalidParameterValue",
locator="VERSION",
exception_text="Grid coordinates x_size * y_size must be less than %d"
% settings.MAX_GRID_COORDS_REQUEST_SIZE,
)

fp = coverage.layer.get_coverage(args)

capabilities = self.wcs_capabilities
fn = coverage.identifier.split(".")[-1] + ".tif"

if args["base_url"]:
capabilities.base_url = args["base_url"]
# Collect garbage
gc.collect()

return capabilities.to_xml()
response = {"fp": fp, "fn": fn}

return response

def handle_wcs_kv(self, args):
if args["request"] == "GetCapabilities":
return self.get_capabilities_wcs(args)

if "version" in args and args["version"] == "1.0.0":
wcs_response = wcs_response_1_0_0
Expand All @@ -101,87 +162,96 @@ def handle_wcs_kv(self, args):
)

if args["request"] == "DescribeCoverage":
return self.describe_coverage_wcs(args, wcs_request, wcs_response)

describe_coverage = wcs_request.DescribeCoverage()
try:
describe_coverage.load_from_kv(args)
describe_coverage.validate()
except Exception:
logger.error(LOAD_FAILURE, exc_info=True)
raise WCSException(exception_text=INVALID_ARGUMENTS)
elif args["request"] == "GetCoverage":
return self.get_coverage_wcs(args, wcs_request)

coverages = [
self.get_coverage_from_id(identifier.value)
for identifier in describe_coverage.identifiers
]
coverage_description = wcs_response.CoverageDescription(coverages=coverages)
raise WCSException(exception_text="KV Request not handled properly: " + str(args))

return coverage_description.to_xml()
def get_capabilities_wms(self, args):
get_capabilities = wms_request_1_3_0.GetCapabilities()
try:
get_capabilities.load_from_kv(args)
get_capabilities.validate()
except Exception:
logger.error(LOAD_FAILURE, exc_info=True)
raise WCSException(exception_text=INVALID_ARGUMENTS)

elif args["request"] == "GetCoverage":
get_coverage = wcs_request.GetCoverage()
try:
get_coverage.load_from_kv(args)
get_coverage.validate()
except Exception:
logger.error(LOAD_FAILURE, exc_info=True)
raise WCSException(exception_text=INVALID_ARGUMENTS)
wms_capabilities = self.wms_capabilities

coverage = self.get_coverage_from_id(get_coverage.identifier.value)
if args["base_url"]:
wms_capabilities.base_url = args["base_url"]
return wms_capabilities.to_xml()

from dateutil.parser import parse
def get_legend_graphic(self, args, wms_request):
get_legend_graphic = wms_request.GetLegendGraphic()
try:
get_legend_graphic.load_from_kv(args)
get_legend_graphic.validate()
except Exception:
logger.error(LOAD_FAILURE, exc_info=True)
raise WCSException(exception_text=INVALID_ARGUMENTS)

coverage = self.get_coverage_from_id(get_legend_graphic.layer.value)

if get_coverage.width == 0:
raise WCSException(
exception_code="InvalidParameterValue",
locator="VERSION",
exception_text="Grid coordinates x_size must be greater than 0",
)
if get_coverage.height == 0:
raise WCSException(
exception_code="InvalidParameterValue",
locator="VERSION",
exception_text="Grid coordinates y_size must be greater than 0",
)
if get_coverage.height * get_coverage.width > settings.MAX_GRID_COORDS_REQUEST_SIZE:
raise WCSException(
exception_code="InvalidParameterValue",
locator="VERSION",
exception_text="Grid coordinates x_size * y_size must be less than %d" % settings.MAX_GRID_COORDS_REQUEST_SIZE,
)
fp = coverage.layer.get_legend_graphic(args)

fn = coverage.identifier.split(".")[-1] + ".png"

fp = coverage.layer.get_coverage(args)
response = {"fp": fp, "fn": fn}
return response

fn = coverage.identifier.split(".")[-1] + ".tif"
def get_map(self, args, wms_request):
get_map = wms_request.GetMap()
try:
get_map.load_from_kv(args)
get_map.validate()
except Exception:
logger.error(LOAD_FAILURE, exc_info=True)
raise WCSException(exception_text=INVALID_ARGUMENTS)

# Collect garbage
gc.collect()
coverage = self.get_coverage_from_id(get_map.layer.value)

response = {"fp": fp, "fn": fn}
# Make sure the request size is correct
if get_map.width == 0:
raise WCSException(
exception_code="InvalidParameterValue",
locator="VERSION",
exception_text="Grid coordinates x_size must be greater than 0",
)
if get_map.height == 0:
raise WCSException(
exception_code="InvalidParameterValue",
locator="VERSION",
exception_text="Grid coordinates y_size must be greater than 0",
)
if get_map.height * get_map.width > settings.MAX_GRID_COORDS_REQUEST_SIZE:
raise WCSException(
exception_code="InvalidParameterValue",
locator="VERSION",
exception_text="Grid coordinates x_size * y_size must be less than %d"
% settings.MAX_GRID_COORDS_REQUEST_SIZE,
)

return response
try:
fp = coverage.layer.get_map(args)
except Exception:
logger.error("Failed to get_map from layer: ", exc_info=True)
raise WCSException(exception_text=INVALID_ARGUMENTS)

raise WCSException(
exception_text="KV Request not handled properly: " + str(args)
)
fn = coverage.identifier.split(".")[-1] + ".png"

def handle_wms_kv(self, args):
if args["request"] == "GetCapabilities":
get_capabilities = wms_request_1_3_0.GetCapabilities()
try:
get_capabilities.load_from_kv(args)
get_capabilities.validate()
except Exception:
logger.error(LOAD_FAILURE, exc_info=True)
raise WCSException(exception_text=INVALID_ARGUMENTS)
# Collect garbage
gc.collect()

wms_capabilities = self.wms_capabilities
response = {"fp": fp, "fn": fn}

if args["base_url"]:
wms_capabilities.base_url = args["base_url"]
return wms_capabilities.to_xml()
return response

def handle_wms_kv(self, args):
if args["request"] == "GetCapabilities":
return self.get_capabilities_wms(args)

if args["request"] == "GetFeatureInfo":
raise WCSException(
Expand All @@ -200,70 +270,9 @@ def handle_wms_kv(self, args):
)

if args["request"].lower() == "getlegendgraphic":
get_legend_graphic = wms_request.GetLegendGraphic()
try:
get_legend_graphic.load_from_kv(args)
get_legend_graphic.validate()
except Exception:
logger.error(LOAD_FAILURE, exc_info=True)
raise WCSException(exception_text=INVALID_ARGUMENTS)

coverage = self.get_coverage_from_id(get_legend_graphic.layer.value)

fp = coverage.layer.get_legend_graphic(args)

fn = coverage.identifier.split(".")[-1] + ".png"

response = {"fp": fp, "fn": fn}
return response
return self.get_legend_graphic(args, wms_request)

if args["request"].lower() == "getmap":
return self.get_map(args, wms_request)

get_map = wms_request.GetMap()
try:
get_map.load_from_kv(args)
get_map.validate()
except Exception:
logger.error(LOAD_FAILURE, exc_info=True)
raise WCSException(exception_text=INVALID_ARGUMENTS)

coverage = self.get_coverage_from_id(get_map.layer.value)

# Make sure the request size is correct
if get_map.width == 0:
raise WCSException(
exception_code="InvalidParameterValue",
locator="VERSION",
exception_text="Grid coordinates x_size must be greater than 0",
)
if get_map.height == 0:
raise WCSException(
exception_code="InvalidParameterValue",
locator="VERSION",
exception_text="Grid coordinates y_size must be greater than 0",
)
if get_map.height * get_map.width > settings.MAX_GRID_COORDS_REQUEST_SIZE:
raise WCSException(
exception_code="InvalidParameterValue",
locator="VERSION",
exception_text="Grid coordinates x_size * y_size must be less than %d" % settings.MAX_GRID_COORDS_REQUEST_SIZE,
)

try:
fp = coverage.layer.get_map(args)
except Exception:
logger.error("Failed to get_map from layer: ", exc_info=True)
raise WCSException(exception_text=INVALID_ARGUMENTS)

fn = coverage.identifier.split(".")[-1] + ".png"

# Collect garbage
gc.collect()

response = {"fp": fp, "fn": fn}

return response

raise WCSException(
exception_text="KV Request not handled properly: " + str(args)
)
raise WCSException(exception_text="KV Request not handled properly: " + str(args))
Loading