Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
1 change: 0 additions & 1 deletion backend/accounts/__init__.py
Original file line number Diff line number Diff line change
@@ -1 +0,0 @@
default_app_config = 'accounts.apps.AccountsConfig'
1 change: 0 additions & 1 deletion backend/csw/__init__.py
Original file line number Diff line number Diff line change
@@ -1 +0,0 @@
default_app_config = 'registry.apps.CswConfig'
4 changes: 2 additions & 2 deletions backend/extras/managers.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
from collections.abc import Mapping
from typing import Any, Optional, Tuple
from typing import Any

from django.db import models
from django.db.models import Case, CharField, Count, F, Q, Value, When
Expand All @@ -14,7 +14,7 @@ class UniqueConstraintDefaultValueManager(models.Manager):
to correctly use get_or_create
"""

def get_or_create(self, defaults: Optional[Mapping[str, Any]] = None, **kwargs: Any) -> Tuple[models.Model, bool]:
def get_or_create(self, defaults: Mapping[str, Any] | None = None, **kwargs: Any) -> tuple[models.Model, bool]:
for constraint in list(filter(lambda constraint: isinstance(constraint, UniqueConstraint), self.model._meta.constraints)):
for field in constraint.fields:
kwargs.update({
Expand Down
2 changes: 1 addition & 1 deletion backend/extras/tasks.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
from logging import Logger
from typing import OrderedDict
from collections import OrderedDict

from celery import Task
from celery.signals import task_postrun, task_prerun
Expand Down
2 changes: 1 addition & 1 deletion backend/extras/viewsets.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
from typing import OrderedDict
from collections import OrderedDict

from camel_converter import to_camel
from django.apps import apps
Expand Down
1 change: 0 additions & 1 deletion backend/notify/__init__.py
Original file line number Diff line number Diff line change
@@ -1 +0,0 @@
default_app_config = 'notify.apps.NotifyConfig'
2 changes: 1 addition & 1 deletion backend/notify/utils.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import json
from typing import OrderedDict
from collections import OrderedDict

from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
Expand Down
1 change: 0 additions & 1 deletion backend/registry/__init__.py
Original file line number Diff line number Diff line change
@@ -1 +0,0 @@
default_app_config = 'registry.apps.RegistryConfig'
7 changes: 3 additions & 4 deletions backend/registry/managers/security.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
from abc import ABC
from typing import Any, List, Tuple
from typing import Any

from django.contrib.auth.models import Group
from django.contrib.gis.db.models import Union
Expand Down Expand Up @@ -34,8 +34,7 @@


class AllowedOgcServiceOperationQuerySet(ABC, models.QuerySet):

def get_entity_identifiers(self, request) -> Tuple[str, List[str]]:
def get_entity_identifiers(self, request) -> tuple[str, list[str]]:
raise NotImplementedError

def filter_by_requested_entity(self, request):
Expand Down Expand Up @@ -204,7 +203,7 @@ def get_with_security_info(self, request: HttpRequest, *args: Any, **kwargs: Any

class WebFeatureServiceSecurityManager(models.Manager.from_queryset(AllowedWebFeatureServiceOperationQuerySet)):

def is_unknown_feature_type(self, service_pk, feature_types: List[str]) -> QuerySet:
def is_unknown_feature_type(self, service_pk, feature_types: list[str]) -> QuerySet:
return ~Exists(self.filter(pk=service_pk, featuretype__identifier__in=feature_types))

def get_allowed_operation_qs(self) -> AllowedWebFeatureServiceOperationQuerySet:
Expand Down
5 changes: 1 addition & 4 deletions backend/registry/models/monitoring.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
from datetime import timedelta
from io import BytesIO
from typing import List

from django.conf import settings
from django.contrib.gis.db import models
Expand Down Expand Up @@ -305,9 +304,7 @@ def run_checks(self, run: WebMapServiceMonitoringRun, *args, **kwargs) -> GetMap


class GetMapProbe(WebMapServiceProbe):
layers: List[Layer] = models.ManyToManyField(
to=Layer
)
layers = models.ManyToManyField(to=Layer)
format = models.ForeignKey(
to=MimeType,
on_delete=models.SET_NULL,
Expand Down
12 changes: 7 additions & 5 deletions backend/registry/models/security.py
Original file line number Diff line number Diff line change
Expand Up @@ -429,11 +429,13 @@ def clean(self):
raise ValidationError(
{
"camouflage": format_html(
_(
"There are configured allowed operation objects. Camouflage can not"
" be disabled. See all allowed operations <a href=%(url)s>here</a>"
)
% {"url": url}
str(
_(
"There are configured allowed operation objects. Camouflage cannot"
" be disabled. See all allowed operations <a href={url}>here</a>."
)
),
url=url,
)
}
)
Expand Down
27 changes: 16 additions & 11 deletions backend/registry/models/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -137,16 +137,9 @@ def get_session_for_request(self) -> Session:

return session

def get_service_type(self):
match(self.__class__.__name__):
case "WebMapService":
return "WMS"
case "WebFeatureService":
return "WFS"
case "CatalogueService":
return "CSW"
case _:
return None
@property
def service_type(self) -> str:
raise NotImplementedError

@property
def get_capabilities_url(self) -> str | bytes:
Expand All @@ -158,7 +151,7 @@ def get_capabilities_url(self) -> str | bytes:
cap_url = update_url_query_params(
cap_url,
{
"SERVICE": self.get_service_type(),
"SERVICE": self.service_type,
"VERSION": OGCServiceVersionEnum(self.version).label,
"REQUEST": "GetCapabilities",
}
Expand Down Expand Up @@ -243,6 +236,10 @@ def root_layer(self):
# Not prefetched → let the DB do the work
return self.layers.get(mptt_parent=None)

@property
def service_type(self) -> str:
return "WMS"

@property
def client(self) -> WebMapServiceClient:
return WebMapServiceClient(
Expand Down Expand Up @@ -314,6 +311,10 @@ class Meta:
indexes = [
] + OgcService.Meta.indexes + AbstractMetadata.Meta.indexes

@property
def service_type(self) -> str:
return "WFS"

@property
def client(self) -> WebFeatureServiceClient:
return WebFeatureServiceClient(
Expand Down Expand Up @@ -344,6 +345,10 @@ class Meta:
indexes = [
] + OgcService.Meta.indexes + AbstractMetadata.Meta.indexes

@property
def service_type(self) -> str:
return "CSW"

@property
def client(self) -> CatalogueServiceClient:
return CatalogueServiceClient(
Expand Down
3 changes: 1 addition & 2 deletions backend/registry/ows_lib/csw/csw.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
import re
from typing import List

from registry.enums.service import HttpMethodEnum, OGCOperationEnum
from registry.ows_lib.client.core import OgcClient
Expand All @@ -25,7 +24,7 @@ def queryable_type_name(self):
else:
return "type"

def get_constraint(self, record_types: List[str]):
def get_constraint(self, record_types: list[str]):
type_name = self.queryable_type_name()
record_types_filters = [
f"<ogc:PropertyIsEqualTo><ogc:PropertyName>{type_name}</ogc:PropertyName><ogc:Literal>{record_type}</ogc:Literal></ogc:PropertyIsEqualTo>" for record_type in record_types]
Expand Down
17 changes: 8 additions & 9 deletions backend/registry/ows_lib/request/ogc_request.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
from typing import Dict, List
from xml.sax.saxutils import unescape

from django.contrib.gis.geos import GEOSGeometry, Polygon
Expand Down Expand Up @@ -32,9 +31,9 @@ class OGCRequest(Request):
def __init__(self, django_request: DjangoRequest = None, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
self._djano_request = django_request
self._ogc_query_params: Dict = {}
self._ogc_query_params: dict = {}
self._bbox: GEOSGeometry = None
self._requested_entities: List[str] = []
self._requested_entities: list[str] = []
self._xml_request = None
self.operation = "unknown"
self.service_version = "unknown"
Expand Down Expand Up @@ -66,13 +65,13 @@ def from_django_request(cls, request: DjangoRequest):
return ogc_request

@property
def requested_entities(self) -> List[str]:
def requested_entities(self) -> list[str]:
"""Returns the list of requested entities

This function analyzes the request and find out which layers or featuretypes, or records are requested.

:return: list of requested layers | list of request featuretypes
:rtype: List[str]
:rtype: list[str]
"""
if not self._requested_entities:
if self.is_wms:
Expand Down Expand Up @@ -273,11 +272,11 @@ def bbox(self) -> GEOSGeometry:
return self._bbox

@property
def ogc_query_params(self) -> Dict:
def ogc_query_params(self) -> dict:
""" Parses the GET parameters into all member variables, which can be found in a ogc request.

:return: all ogc query parameters
:rtype: Dict
:rtype: dict
"""
if not self._ogc_query_params:
query_keys = ["SERVICE", "REQUEST", "LAYERS", "BBOX", "VERSION", "FORMAT",
Expand Down Expand Up @@ -410,12 +409,12 @@ def content(self) -> bytes:
encoding="UTF-8"
)

def secure_get_feature_request(self, security_info_per_feature_type: List[dict]):
def secure_get_feature_request(self, security_info_per_feature_type: list[dict]):
"""

Parameters
----------
security_info_per_feature_type: List[dict]
security_info_per_feature_type: list[dict]
An array of JSON objects, one per requested feature type, with the
following structure:

Expand Down
27 changes: 10 additions & 17 deletions backend/registry/ows_lib/request/utils.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
import urllib.parse
from typing import Dict, List, Union

from django.contrib.gis.gdal import SpatialReference
from django.contrib.gis.geos import GEOSGeometry, Polygon
Expand All @@ -11,7 +10,7 @@
MissingServiceParam)


def update_queryparams(url: str, params: Dict):
def update_queryparams(url: str, params: dict):
"""Helper function to update query paramase inside an existing url with trailing query params"""
url_parts = urllib.parse.urlparse(url)
query = dict(urllib.parse.parse_qsl(url_parts.query))
Expand Down Expand Up @@ -169,20 +168,18 @@ def construct_polygon_from_bbox_query_param(get_dict) -> GEOSGeometry:
return _construct_polygon_from_bbox_query_param_for_wfs(get_dict=get_dict)


def get_requested_layers(params: Dict) -> List[str]:
def get_requested_layers(params: dict) -> list[str]:
"""Filters the given params by requested layers

:param params: all query parameters
:type params: Dict
:type params: dict
:return: the requested layers from the query params
:rtype: List[str]
:rtype: list[str]
"""
return list(filter(None, params.get("LAYERS", params.get("layers", "")).split(",")))


def get_requested_feature_types(
params: Union[Dict[str, str], etree._Element]
) -> List[str]:
def get_requested_feature_types(params: dict[str, str] | etree._Element) -> list[str]:
"""
Extract requested feature types from query parameters or an XML element.

Expand All @@ -200,7 +197,7 @@ def get_requested_feature_types(

# --- Case 2: XML element ---
if isinstance(params, etree._Element):
feature_types: List[str] = []
feature_types: list[str] = []

# map default namespace to 'wfs' if None
nsmap = {k if k is not None else "wfs": v for k,
Expand All @@ -217,13 +214,10 @@ def get_requested_feature_types(

return feature_types

raise TypeError(
"params must be either Dict[str, str] or lxml.etree._Element")
raise TypeError("params must be either dict[str, str] or lxml.etree._Element")


def get_requested_records(
params: Union[Dict[str, str], etree._Element]
) -> List[str]:
def get_requested_records(params: dict[str, str] | etree._Element) -> list[str]:
"""
Extract requested record IDs from query parameters or a CSW XML element.

Expand All @@ -241,7 +235,7 @@ def get_requested_records(

# --- Case 2: XML element ---
if isinstance(params, etree._Element):
record_ids: List[str] = []
record_ids: list[str] = []

# auto-detect namespaces
nsmap = {k if k is not None else "default": v for k,
Expand All @@ -255,5 +249,4 @@ def get_requested_records(

return record_ids

raise TypeError(
"params must be either Dict[str, str] or lxml.etree._Element")
raise TypeError("params must be either dict[str, str] or lxml.etree._Element")
6 changes: 2 additions & 4 deletions backend/registry/ows_lib/wfs/wfs.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,3 @@
from typing import List

from lxml.etree import _Element
from registry.enums.service import HttpMethodEnum, OGCOperationEnum
from registry.ows_lib.client.core import OgcClient
Expand Down Expand Up @@ -28,8 +26,8 @@ def output_format_qp(self):

def describe_feature_type_request(
self,
type_names: List[str],
output_format: List[str]) -> Request:
type_names: list[str],
output_format: list[str]) -> Request:

params = {
"VERSION": self.service_version,
Expand Down
Loading