Code Style

Author: Ornella33

server/app/adapter/__init__.py

@@ -1 +1 @@
-from .jsonization import *
+from .jsonization import *

server/app/adapter/jsonization.py

@@ -1 +1 @@
-from .local_file import *
+from .local_file import *

server/app/backend/__init__.py

@@ -1,18 +1,16 @@
-from typing import Iterator, Dict, Type, Union
-import logging
+import hashlib
 import json
+import logging
 import os
-import hashlib
 import threading
 import weakref
+from typing import Dict, Iterator, Type, Union
 
-from app.model import AssetAdministrationShellDescriptor, SubmodelDescriptor
 from basyx.aas import model
 from basyx.aas.model import provider as sdk_provider
 
-from app.model import descriptor
 from app.adapter import jsonization
-
+from app.model import AssetAdministrationShellDescriptor, SubmodelDescriptor, descriptor
 
 logger = logging.getLogger(__name__)
 
@@ -31,6 +29,7 @@ class LocalFileDescriptorStore(sdk_provider.AbstractObjectStore[model.Identifier
     An ObjectStore implementation for :class:`~app.model.descriptor.Descriptor` BaSyx Python SDK objects backed
     by a local file based local backend
     """
+
     def __init__(self, directory_path: str):
         """
         Initializer of class LocalFileDescriptorStore
@@ -44,8 +43,9 @@ def __init__(self, directory_path: str):
         # local replication of each object is kept in the application and retrieving an object from the store always
         # returns the **same** (not only equal) object. Still, objects are forgotten, when they are not referenced
         # anywhere else to save memory.
-        self._object_cache: weakref.WeakValueDictionary[model.Identifier, _DESCRIPTOR_TYPE] \
-            = weakref.WeakValueDictionary()
+        self._object_cache: weakref.WeakValueDictionary[model.Identifier, _DESCRIPTOR_TYPE] = (
+            weakref.WeakValueDictionary()
+        )
         self._object_cache_lock = threading.Lock()
 
     def check_directory(self, create=False):
@@ -107,9 +107,7 @@ def add(self, x: _DESCRIPTOR_TYPE) -> None:
             # Usually, we don't need to serialize the modelType, since during HTTP requests, we know exactly if this
             # is an AASDescriptor or SubmodelDescriptor. However, here we cannot distinguish them, so to deserialize
             # them successfully, we hack the `modelType` into the JSON.
-            serialized = json.loads(
-                json.dumps(x, cls=jsonization.ServerAASToJsonEncoder)
-            )
+            serialized = json.loads(json.dumps(x, cls=jsonization.ServerAASToJsonEncoder))
             serialized["modelType"] = DESCRIPTOR_TYPE_TO_STRING[type(x)]
             json.dump(serialized, file, indent=4)
             with self._object_cache_lock:
@@ -173,4 +171,4 @@ def _transform_id(identifier: model.Identifier) -> str:
         """
         Helper method to represent an ASS Identifier as a string to be used as Local file document id
         """
-        return hashlib.sha256(identifier.encode("utf-8")).hexdigest()
+        return hashlib.sha256(identifier.encode("utf-8")).hexdigest()

server/app/backend/local_file.py

@@ -1,4 +1,4 @@
 from .base import *
-from .registry import *
 from .discovery import *
-from .repository import *
+from .registry import *
+from .repository import *

server/app/interfaces/__init__.py

@@ -10,27 +10,24 @@
 import io
 import itertools
 import json
-from typing import Iterable, Type, Iterator, Tuple, Optional, List, Union, Dict, Callable, TypeVar, Any
+from typing import Any, Callable, Dict, Iterable, Iterator, List, Optional, Tuple, Type, TypeVar, Union
 
 import werkzeug.exceptions
 import werkzeug.routing
 import werkzeug.utils
+from basyx.aas import model
+from basyx.aas.adapter._generic import XML_NS_MAP
+from basyx.aas.adapter.xml import XMLConstructables, read_aas_xml_element, xml_serialization
+from basyx.aas.model import AbstractObjectStore
 from lxml import etree
-from werkzeug import Response, Request
-from werkzeug.exceptions import NotFound, BadRequest
+from werkzeug import Request, Response
+from werkzeug.exceptions import BadRequest, NotFound
 from werkzeug.routing import MapAdapter
 
 import app.model
-from basyx.aas import model
-from basyx.aas.adapter._generic import XML_NS_MAP
-from app.adapter import ServerStrictStrippedAASFromJsonDecoder, ServerStrictAASFromJsonDecoder, ServerAASToJsonEncoder
-from basyx.aas.adapter.xml import xml_serialization, XMLConstructables, read_aas_xml_element
-from basyx.aas.model import AbstractObjectStore
+from app.adapter import ServerAASToJsonEncoder, ServerStrictAASFromJsonDecoder, ServerStrictStrippedAASFromJsonDecoder
+from app.model import AssetAdministrationShellDescriptor, AssetLink, SubmodelDescriptor
 from app.util.converters import base64url_decode
-from app.model import AssetLink, AssetAdministrationShellDescriptor, SubmodelDescriptor
-from app.model.provider import _DESCRIPTOR_TYPE
-
-
 
 T = TypeVar("T")
 
@@ -48,13 +45,19 @@ def __str__(self):
 
 
 class Message:
-    def __init__(self, code: str, text: str, message_type: MessageType = MessageType.UNDEFINED,
-                 timestamp: Optional[datetime.datetime] = None):
+    def __init__(
+        self,
+        code: str,
+        text: str,
+        message_type: MessageType = MessageType.UNDEFINED,
+        timestamp: Optional[datetime.datetime] = None,
+    ):
         self.code: str = code
         self.text: str = text
         self.message_type: MessageType = message_type
-        self.timestamp: datetime.datetime = timestamp if timestamp is not None \
-            else datetime.datetime.now(datetime.timezone.utc)
+        self.timestamp: datetime.datetime = (
+            timestamp if timestamp is not None else datetime.datetime.now(datetime.timezone.utc)
+        )
 
 
 class Result:
@@ -70,8 +73,9 @@ def __init__(self, success: bool, messages: Optional[List[Message]] = None):
 
 class APIResponse(abc.ABC, Response):
     @abc.abstractmethod
-    def __init__(self, obj: Optional[ResponseData] = None, cursor: Optional[int] = None,
-                 stripped: bool = False, *args, **kwargs):
+    def __init__(
+        self, obj: Optional[ResponseData] = None, cursor: Optional[int] = None, stripped: bool = False, *args, **kwargs
+    ):
         super().__init__(*args, **kwargs)
         if obj is None:
             self.status_code = 204
@@ -91,14 +95,9 @@ def serialize(self, obj: ResponseData, cursor: Optional[int], stripped: bool) ->
         if cursor is None:
             data = obj
         else:
-            data = {
-                "paging_metadata": {"cursor": str(cursor)},
-                "result": obj
-            }
+            data = {"paging_metadata": {"cursor": str(cursor)}, "result": obj}
         return json.dumps(
-            data,
-            cls=StrippedResultToJsonEncoder if stripped else ResultToJsonEncoder,
-            separators=(",", ":")
+            data, cls=StrippedResultToJsonEncoder if stripped else ResultToJsonEncoder, separators=(",", ":")
         )
 
 
@@ -166,18 +165,15 @@ def __init__(self, *args, content_type="text/xml", **kwargs):
 class ResultToJsonEncoder(ServerAASToJsonEncoder):
     @classmethod
     def _result_to_json(cls, result: Result) -> Dict[str, object]:
-        return {
-            "success": result.success,
-            "messages": result.messages
-        }
+        return {"success": result.success, "messages": result.messages}
 
     @classmethod
     def _message_to_json(cls, message: Message) -> Dict[str, object]:
         return {
             "messageType": message.message_type,
             "text": message.text,
             "code": message.code,
-            "timestamp": message.timestamp.isoformat()
+            "timestamp": message.timestamp.isoformat(),
         }
 
     def default(self, obj: object) -> object:
@@ -204,8 +200,8 @@ def __call__(self, environ, start_response) -> Iterable[bytes]:
 
     @classmethod
     def _get_slice(cls, request: Request, iterator: Iterable[T]) -> Tuple[Iterator[T], int]:
-        limit_str = request.args.get('limit', default="10")
-        cursor_str = request.args.get('cursor', default="1")
+        limit_str = request.args.get("limit", default="10")
+        cursor_str = request.args.get("cursor", default="1")
         try:
             limit, cursor = int(limit_str), int(cursor_str) - 1  # cursor is 1-indexed
             if limit < 0 or cursor < 0:
@@ -238,27 +234,31 @@ def get_response_type(request: Request) -> Type[APIResponse]:
         response_types: Dict[str, Type[APIResponse]] = {
             "application/json": JsonResponse,
             "application/xml": XmlResponse,
-            "text/xml": XmlResponseAlt
+            "text/xml": XmlResponseAlt,
         }
         if len(request.accept_mimetypes) == 0 or request.accept_mimetypes.best in (None, "*/*"):
             return JsonResponse
         mime_type = request.accept_mimetypes.best_match(response_types)
         if mime_type is None:
-            raise werkzeug.exceptions.NotAcceptable("This server supports the following content types: "
-                                                    + ", ".join(response_types.keys()))
+            raise werkzeug.exceptions.NotAcceptable(
+                "This server supports the following content types: " + ", ".join(response_types.keys())
+            )
         return response_types[mime_type]
 
     @staticmethod
-    def http_exception_to_response(exception: werkzeug.exceptions.HTTPException, response_type: Type[APIResponse]) \
-            -> APIResponse:
+    def http_exception_to_response(
+        exception: werkzeug.exceptions.HTTPException, response_type: Type[APIResponse]
+    ) -> APIResponse:
         headers = exception.get_headers()
         location = exception.get_response().location
         if location is not None:
             headers.append(("Location", location))
         if exception.code and exception.code >= 400:
-            message = Message(type(exception).__name__,
-                              exception.description if exception.description is not None else "",
-                              MessageType.ERROR)
+            message = Message(
+                type(exception).__name__,
+                exception.description if exception.description is not None else "",
+                MessageType.ERROR,
+            )
             result = Result(False, [message])
         else:
             result = Result(False)
@@ -273,13 +273,13 @@ def _get_all_obj_of_type(self, type_: Type[T]) -> Iterator[T]:
             if isinstance(obj, type_):
                 yield obj
 
-    def _get_obj_ts(self, identifier: model.Identifier, type_: Type[T]) \
-            -> T:
+    def _get_obj_ts(self, identifier: model.Identifier, type_: Type[T]) -> T:
         identifiable = self.object_store.get(identifier)
         if not isinstance(identifiable, type_):
             raise NotFound(f"No {type_.__name__} with {identifier} found!")
         return identifiable
 
+
 class HTTPApiDecoder:
     # these are the types we can construct (well, only the ones we need)
     type_constructables_map = {
@@ -313,8 +313,9 @@ def assert_type(cls, obj: object, type_: Type[T]) -> T:
     @classmethod
     def json_list(cls, data: Union[str, bytes], expect_type: Type[T], stripped: bool, expect_single: bool) -> List[T]:
         cls.check_type_support(expect_type)
-        decoder: Type[ServerStrictAASFromJsonDecoder] = ServerStrictStrippedAASFromJsonDecoder if stripped \
-            else ServerStrictAASFromJsonDecoder
+        decoder: Type[ServerStrictAASFromJsonDecoder] = (
+            ServerStrictStrippedAASFromJsonDecoder if stripped else ServerStrictAASFromJsonDecoder
+        )
         try:
             parsed = json.loads(data, cls=decoder)
             if isinstance(parsed, list) and expect_single:
@@ -371,8 +372,9 @@ def xml(cls, data: bytes, expect_type: Type[T], stripped: bool) -> T:
         cls.check_type_support(expect_type)
         try:
             xml_data = io.BytesIO(data)
-            rv = read_aas_xml_element(xml_data, cls.type_constructables_map[expect_type],
-                                      stripped=stripped, failsafe=False)
+            rv = read_aas_xml_element(
+                xml_data, cls.type_constructables_map[expect_type], stripped=stripped, failsafe=False
+            )
         except (KeyError, ValueError) as e:
             # xml deserialization creates an error chain. since we only return one error, return the root cause
             f: BaseException = e
@@ -397,8 +399,8 @@ def request_body(cls, request: Request, expect_type: Type[T], stripped: bool) ->
 
         if request.mimetype not in valid_content_types:
             raise werkzeug.exceptions.UnsupportedMediaType(
-                f"Invalid content-type: {request.mimetype}! Supported types: "
-                + ", ".join(valid_content_types))
+                f"Invalid content-type: {request.mimetype}! Supported types: " + ", ".join(valid_content_types)
+            )
 
         if request.mimetype == "application/json":
             return cls.json(request.get_data(), expect_type, stripped)