diff --git a/backend/accounts/__init__.py b/backend/accounts/__init__.py
index 831982319..e69de29bb 100644
--- a/backend/accounts/__init__.py
+++ b/backend/accounts/__init__.py
@@ -1 +0,0 @@
-default_app_config = 'accounts.apps.AccountsConfig'
diff --git a/backend/csw/__init__.py b/backend/csw/__init__.py
index bcd495512..e69de29bb 100644
--- a/backend/csw/__init__.py
+++ b/backend/csw/__init__.py
@@ -1 +0,0 @@
-default_app_config = 'registry.apps.CswConfig'
diff --git a/backend/extras/managers.py b/backend/extras/managers.py
index f48748e33..34f45b31b 100644
--- a/backend/extras/managers.py
+++ b/backend/extras/managers.py
@@ -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
@@ -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({
diff --git a/backend/extras/tasks.py b/backend/extras/tasks.py
index 87f5d472f..950f67305 100644
--- a/backend/extras/tasks.py
+++ b/backend/extras/tasks.py
@@ -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
diff --git a/backend/extras/viewsets.py b/backend/extras/viewsets.py
index 6dde4a851..778df8982 100644
--- a/backend/extras/viewsets.py
+++ b/backend/extras/viewsets.py
@@ -1,4 +1,4 @@
-from typing import OrderedDict
+from collections import OrderedDict
from camel_converter import to_camel
from django.apps import apps
diff --git a/backend/notify/__init__.py b/backend/notify/__init__.py
index c08abdf99..e69de29bb 100644
--- a/backend/notify/__init__.py
+++ b/backend/notify/__init__.py
@@ -1 +0,0 @@
-default_app_config = 'notify.apps.NotifyConfig'
diff --git a/backend/notify/utils.py b/backend/notify/utils.py
index c1356a578..e540a98a7 100644
--- a/backend/notify/utils.py
+++ b/backend/notify/utils.py
@@ -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
diff --git a/backend/registry/__init__.py b/backend/registry/__init__.py
index 905d64bce..e69de29bb 100644
--- a/backend/registry/__init__.py
+++ b/backend/registry/__init__.py
@@ -1 +0,0 @@
-default_app_config = 'registry.apps.RegistryConfig'
diff --git a/backend/registry/managers/security.py b/backend/registry/managers/security.py
index 23b3b9a24..867fbf32e 100644
--- a/backend/registry/managers/security.py
+++ b/backend/registry/managers/security.py
@@ -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
@@ -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):
@@ -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:
diff --git a/backend/registry/models/monitoring.py b/backend/registry/models/monitoring.py
index 65bc7f48a..ead69357a 100644
--- a/backend/registry/models/monitoring.py
+++ b/backend/registry/models/monitoring.py
@@ -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
@@ -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,
diff --git a/backend/registry/models/security.py b/backend/registry/models/security.py
index 87b59bca9..109bf7820 100644
--- a/backend/registry/models/security.py
+++ b/backend/registry/models/security.py
@@ -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 here"
- )
- % {"url": url}
+ str(
+ _(
+ "There are configured allowed operation objects. Camouflage cannot"
+ " be disabled. See all allowed operations here."
+ )
+ ),
+ url=url,
)
}
)
diff --git a/backend/registry/models/service.py b/backend/registry/models/service.py
index ae9618479..ce39a83b1 100644
--- a/backend/registry/models/service.py
+++ b/backend/registry/models/service.py
@@ -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:
@@ -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",
}
@@ -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(
@@ -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(
@@ -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(
diff --git a/backend/registry/ows_lib/csw/csw.py b/backend/registry/ows_lib/csw/csw.py
index 105279964..579e8ce42 100644
--- a/backend/registry/ows_lib/csw/csw.py
+++ b/backend/registry/ows_lib/csw/csw.py
@@ -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
@@ -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"{type_name}{record_type}" for record_type in record_types]
diff --git a/backend/registry/ows_lib/request/ogc_request.py b/backend/registry/ows_lib/request/ogc_request.py
index 22a08c2cf..c7e7a675b 100644
--- a/backend/registry/ows_lib/request/ogc_request.py
+++ b/backend/registry/ows_lib/request/ogc_request.py
@@ -1,4 +1,3 @@
-from typing import Dict, List
from xml.sax.saxutils import unescape
from django.contrib.gis.geos import GEOSGeometry, Polygon
@@ -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"
@@ -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:
@@ -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",
@@ -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:
diff --git a/backend/registry/ows_lib/request/utils.py b/backend/registry/ows_lib/request/utils.py
index 624f2fdf7..cc04aa554 100644
--- a/backend/registry/ows_lib/request/utils.py
+++ b/backend/registry/ows_lib/request/utils.py
@@ -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
@@ -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))
@@ -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.
@@ -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,
@@ -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.
@@ -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,
@@ -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")
diff --git a/backend/registry/ows_lib/wfs/wfs.py b/backend/registry/ows_lib/wfs/wfs.py
index 388cd68c3..6fe98ac34 100644
--- a/backend/registry/ows_lib/wfs/wfs.py
+++ b/backend/registry/ows_lib/wfs/wfs.py
@@ -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
@@ -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,
diff --git a/backend/registry/ows_lib/wms/wms.py b/backend/registry/ows_lib/wms/wms.py
index a253328aa..beed99c26 100644
--- a/backend/registry/ows_lib/wms/wms.py
+++ b/backend/registry/ows_lib/wms/wms.py
@@ -1,5 +1,3 @@
-from typing import Dict, List, Tuple
-
from registry.enums.service import HttpMethodEnum, OGCOperationEnum
from registry.ows_lib.client.core import OgcClient
from registry.ows_lib.request.utils import update_queryparams
@@ -23,29 +21,29 @@ def get_feature_info_operation_name(self):
def get_map_request(
self,
- layers: List[str],
- styles: List[str],
+ layers: list[str],
+ styles: list[str],
crs: str,
- bbox: Tuple[float, float, float, float],
+ bbox: tuple[float, float, float, float],
width: int,
height: int,
format: str,
transparent: bool = None,
bgcolor: int = 0xFFFFFF,
exceptions: str = "xml",
- time: List[str] = None,
+ time: list[str] = None,
elevation: float = None) -> Request:
"""Constructs a GetMap request to use for requesting
:param layers: The name of layers which shall be requested
- :type layers: List[str]
+ :type layers: list[str]
:param styles: The styles of the layers which shall be requested
- :type styles: List[str]
+ :type styles: list[str]
:param crs: the reference system which shall be used
:type crs: str
- :param bbox: the bounding box
- :type bbox: Tuple[float, float, float, float]
- :param width: the pixel width
+ :param bbox: the bounding box
+ :type bbox: tuple[float, float, float, float]
+ :param width: the pixel width
:type width: int
:param height: the pixel height
:type height: int
@@ -58,7 +56,7 @@ def get_map_request(
:param exceptions: the exception format which shall be used by the server, defaults to "xml"
:type exceptions: str, optional
:param time: the time value or range for map data, defaults to None
- :type time: List[str], optional
+ :type time: list[str], optional
:param elevation: _description_, defaults to None
:type elevation: float, optional
:return: the constructed get map request object
@@ -103,7 +101,7 @@ def get_map_request(
def get_feature_info_request(
self,
get_map_request: Request,
- query_layers: List[str],
+ query_layers: list[str],
info_format: str,
i: int,
j: int,
@@ -114,7 +112,7 @@ def get_feature_info_request(
:param get_map_request: The GetMap request where this request shall based on
:type get_map_request: requests.Request
:param query_layers: The list of layers for that the feature info shall be requested
- :type query_layers: List[str]
+ :type query_layers: list[str]
:param info_format: The concrete format of the response
:type info_format: str
:param i: the x value of the x/y point tuple
@@ -129,7 +127,7 @@ def get_feature_info_request(
:rtype: requests.Request
"""
- params: Dict = get_map_request.params
+ params: dict = get_map_request.params
get_feature_info_params = {
"VERSION": self.service_version,
diff --git a/backend/registry/proxy/mixins.py b/backend/registry/proxy/mixins.py
index edb1a1ae2..77657095a 100644
--- a/backend/registry/proxy/mixins.py
+++ b/backend/registry/proxy/mixins.py
@@ -1,5 +1,6 @@
import datetime
import re
+from functools import cached_property
from io import BytesIO
from django.contrib.auth import get_user_model
@@ -39,8 +40,6 @@ class OgcServiceProxyView(View):
"""
bbox = None
start_time = None
- _service = None
- _remote_service = None
@property
def is_get_request(self) -> bool:
@@ -50,28 +49,24 @@ def is_get_request(self) -> bool:
def is_post_request(self) -> bool:
return self.request.method == "POST"
- @property
+ @cached_property
def service(self) -> OgcService:
- if not self._service:
- try:
- qs = self.service_cls.security.prefetch_whole_service()
- self._service = self.service_cls.security.prepare_with_security_info(self.ogc_request, qs=qs).get(
- pk=self.kwargs.get("pk")
- )
- except ObjectDoesNotExist:
- raise Http404
- return self._service
+ try:
+ qs = self.service_cls.security.prefetch_whole_service()
+ return self.service_cls.security.prepare_with_security_info(self.ogc_request, qs=qs).get(
+ pk=self.kwargs.get("pk")
+ )
+ except ObjectDoesNotExist:
+ raise Http404
@property
def service_cls(self):
raise ImproperlyConfigured(
"you need to setup the proxy class with the corretc 'service_cls' property.")
- @property
+ @cached_property
def remote_service(self) -> OgcClient:
- if not self._remote_service:
- self._remote_service = self.service.client
- return self._remote_service
+ return self.service.client
def analyze_request(self):
"""hook method to do adittional stuff in child classes"""
diff --git a/backend/registry/proxy/wfs_proxy.py b/backend/registry/proxy/wfs_proxy.py
index 77fe81aa1..cb8888c73 100644
--- a/backend/registry/proxy/wfs_proxy.py
+++ b/backend/registry/proxy/wfs_proxy.py
@@ -1,6 +1,5 @@
-
-
import json
+from functools import cached_property
from django.contrib.gis.geos import GEOSGeometry
from django.utils.decorators import method_decorator
@@ -29,39 +28,38 @@ class WebFeatureServiceProxy(OgcServiceProxyView):
service_cls = WebFeatureService
- @property
+ @cached_property
def service(self) -> WebFeatureService:
- if not self._service:
- self._service = super().service
-
- # The security manager of server model does provide security_info_per_feature_type attribute as json encoded object.
- # To use the information in common django/python way, we convert the given json object to dict/geos objects.
- _security_info_per_feature_type = []
-
- for security_info in self._service.security_info_per_feature_type:
- polygon_dict = security_info.get("allowed_area_union")
- geometry = GEOSGeometry(json.dumps(polygon_dict))
- # The given geometry object has always the x/y representation for there coordinates
- # from geoserver docs: https://docs.geoserver.org/stable/en/user/services/wfs/axis_order.html
- # WFS 1.0.0: Provides geographic coordinates in east/north and may not be trusted to respect the EPSG definition axis order.
- # So for WFS 1.0.0 we don't need anything to do. It is interpreted as x/y
- if self.ogc_request.service_version != "1.0.0":
- # All other versions of wfs uses the epsg definition of axis ordering.
- # WFS 1.1.0: Respects the axis order defined by the EPSG definition.
- # WFS 2.0.0: Respects the axis order defined by the EPSG definition.
- # So we need to request the epsg registry and adjust the axis order if needed.
- geometry = adjust_axis_order(geometry)
- _security_info_per_feature_type.append(
- {
- "type_name": security_info.get("type_name"),
- "geometry_property_name": security_info.get("geometry_property_name"),
- "allowed_area_union": geometry
- }
- )
-
- self._service.security_info_per_feature_type = _security_info_per_feature_type
-
- return self._service
+ service = super().service
+
+ # The security manager of server model does provide security_info_per_feature_type attribute as json encoded object.
+ # To use the information in common django/python way, we convert the given json object to dict/geos objects.
+ security_info_per_feature_type = []
+
+ for security_info in service.security_info_per_feature_type:
+ polygon_dict = security_info.get("allowed_area_union")
+ geometry = GEOSGeometry(json.dumps(polygon_dict))
+ # The given geometry object has always the x/y representation for there coordinates
+ # from geoserver docs: https://docs.geoserver.org/stable/en/user/services/wfs/axis_order.html
+ # WFS 1.0.0: Provides geographic coordinates in east/north and may not be trusted to respect the EPSG definition axis order.
+ # So for WFS 1.0.0 we don't need anything to do. It is interpreted as x/y
+ if self.ogc_request.service_version != "1.0.0":
+ # All other versions of wfs uses the epsg definition of axis ordering.
+ # WFS 1.1.0: Respects the axis order defined by the EPSG definition.
+ # WFS 2.0.0: Respects the axis order defined by the EPSG definition.
+ # So we need to request the epsg registry and adjust the axis order if needed.
+ geometry = adjust_axis_order(geometry)
+ security_info_per_feature_type.append(
+ {
+ "type_name": security_info.get("type_name"),
+ "geometry_property_name": security_info.get("geometry_property_name"),
+ "allowed_area_union": geometry,
+ }
+ )
+
+ service.security_info_per_feature_type = security_info_per_feature_type
+
+ return service
@property
def remote_service(self) -> WebFeatureServiceClient:
diff --git a/backend/registry/querys/service.py b/backend/registry/querys/service.py
index cd72eeef2..ea7387d7d 100644
--- a/backend/registry/querys/service.py
+++ b/backend/registry/querys/service.py
@@ -303,6 +303,8 @@ def with_inherited_attributes_cte(self, request=None):
class LayerPrefetch(Prefetch):
+ def __init__(self, request=None, to_attr=None):
+ super().__init__("layers", queryset=self.get_queryset(request), to_attr=to_attr)
def get_queryset(self, request=None):
from registry.models.service import Layer
@@ -340,14 +342,6 @@ def get_queryset(self, request=None):
return qs
- def __init__(self, request=None, *args, **kwargs):
- super().__init__(
- "layers",
- queryset=self.get_queryset(request),
- *args,
- **kwargs
- )
-
class RequestBasedPrefetch(Prefetch):
def __init__(self,
diff --git a/backend/registry/serializers/service.py b/backend/registry/serializers/service.py
index ce1183bbb..9610b5434 100644
--- a/backend/registry/serializers/service.py
+++ b/backend/registry/serializers/service.py
@@ -1,5 +1,3 @@
-from typing import Dict, List
-
from accounts.models.groups import Organization
from accounts.serializers.users import UserSerializer
from django.utils.translation import gettext_lazy as _
@@ -189,7 +187,7 @@ class Meta:
fields = "__all__"
def _convert_dict_to_object(self, objs, Model, defaults={}):
- if objs and isinstance(objs, List) and isinstance(objs[0], Dict):
+ if objs and isinstance(objs, list) and isinstance(objs[0], dict):
return [Model(**defaults, **obj) for obj in objs]
return objs
diff --git a/backend/registry/tasks/monitoring.py b/backend/registry/tasks/monitoring.py
index d3bb89bf1..c6d75cecd 100644
--- a/backend/registry/tasks/monitoring.py
+++ b/backend/registry/tasks/monitoring.py
@@ -1,5 +1,3 @@
-from typing import List
-
from celery import chain, group, shared_task
from django.utils import timezone
from registry.models.monitoring import (GetCapabilitiesProbe, GetMapProbe,
@@ -36,9 +34,9 @@ def finish_run(self, _group_result, run_pk, *args, **kwargs):
def run_wms_monitoring(self, run_pk=None, *args, **kwargs):
run = WebMapServiceMonitoringRun.objects.select_related(
"setting").get(pk=run_pk)
- get_capabilitites_probes: List[GetCapabilitiesProbe] = run.setting.registry_getcapabilitiesprobes.all(
+ get_capabilitites_probes: list[GetCapabilitiesProbe] = run.setting.registry_getcapabilitiesprobes.all(
)
- get_map_probes: List[GetMapProbe] = run.setting.registry_getcapabilitiesprobes.all(
+ get_map_probes: list[GetMapProbe] = run.setting.registry_getcapabilitiesprobes.all(
)
tasks = []
diff --git a/backend/system/logging/util.py b/backend/system/logging/util.py
index 138b48b9f..b3a0bba06 100644
--- a/backend/system/logging/util.py
+++ b/backend/system/logging/util.py
@@ -1,5 +1,5 @@
import re
-from typing import Any, Dict
+from typing import Any
import sqlparse
@@ -20,12 +20,12 @@ def interpolate_sql(sql, params):
return f"{sql} -- [Interpolation failed: {e}]"
-def parse_rfc5424_message(msg: str) -> Dict[str, Any]:
+def parse_rfc5424_message(msg: str) -> dict[str, Any]:
"""
Zerlegt eine durch RFC5424Formatter erzeugte Logzeile zurück in ihre Einzelteile.
Gibt Dict mit header-Feldern, structured_data und message zurück.
"""
- result: Dict[str, Any] = {}
+ result: dict[str, Any] = {}
# Regex grob: "1 TIMESTAMP HOST APP PROCID - [SD...] MESSAGE"
# Structured Data kann mehrfach vorkommen → separate Extraktion
@@ -89,7 +89,7 @@ def parse_rfc5424_message(msg: str) -> Dict[str, Any]:
return result
-def format_structured_data(structured_data: Dict):
+def format_structured_data(structured_data: dict):
extra_sd_elements = []
for sdid, key_value_pairs in structured_data.items():
sd = f"[{sdid}"
diff --git a/backend/tests/django/notify/test_signals.py b/backend/tests/django/notify/test_signals.py
index 21e120df2..58351e8db 100644
--- a/backend/tests/django/notify/test_signals.py
+++ b/backend/tests/django/notify/test_signals.py
@@ -1,5 +1,5 @@
import asyncio
-from typing import OrderedDict
+from collections import OrderedDict
from unittest import skip
from asgiref.sync import async_to_sync