minor polishing

Author: whummer

typedb/localstack_typedb/extension.py

@@ -5,6 +5,7 @@
 from localstack.utils.docker_utils import DOCKER_CLIENT
 from localstack_typedb.utils.docker import ProxiedDockerContainerExtension
 from rolo import Request
+from werkzeug.datastructures import Headers
 
 # environment variable for user-defined command args to pass to TypeDB
 ENV_CMD_FLAGS = "TYPEDB_FLAGS"
@@ -43,6 +44,15 @@ def _get_image_command(self) -> list[str]:
         image_command = result["Config"]["Cmd"]
         return image_command
 
+    def should_proxy_request(self, headers: Headers) -> bool:
+        # determine if this is a gRPC request targeting TypeDB
+        content_type = headers.get("content-type") or ""
+        req_path = headers.get(":path") or ""
+        is_typedb_grpc_request = (
+            "grpc" in content_type and "/typedb.protocol.TypeDB" in req_path
+        )
+        return is_typedb_grpc_request
+
     def request_to_port_router(self, request: Request) -> int:
         # TODO add REST API / gRPC routing based on request
         return 1729

typedb/localstack_typedb/utils/docker.py

@@ -19,7 +19,6 @@
 from rolo import route
 from rolo.proxy import Proxy
 from rolo.routing import RuleAdapter, WithHost
-from werkzeug.datastructures import Headers
 
 LOG = logging.getLogger(__name__)
 logging.getLogger("localstack_typedb").setLevel(
@@ -109,15 +108,6 @@ def _get_container_name(self) -> str:
         name = re.sub(r"\W", "-", name)
         return name
 
-    def should_proxy_request(self, headers: Headers) -> bool:
-        # determine if this is a gRPC request targeting TypeDB
-        content_type = headers.get("content-type") or ""
-        req_path = headers.get(":path") or ""
-        is_typedb_grpc_request = (
-            "grpc" in content_type and "/typedb.protocol.TypeDB" in req_path
-        )
-        return is_typedb_grpc_request
-
     @cache
     def start_container(self) -> None:
         container_name = self._get_container_name()