Handle all get requests via a single API call

schunk_egu_egk_gripper_dummy/main.py

@@ -1,6 +1,6 @@
 from src.dummy import Dummy
 
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
 from typing import Optional
 from pydantic import BaseModel
@@ -28,20 +28,10 @@ class Message(BaseModel):
 
 @server.put("/")
 async def put(msg: Message):
-    dummy.process(msg.message)
+    print(msg)
     return True
 
 
-@server.get("/adi/info.json")
-async def get_info():
-    return dummy.get_info()
-
-
-@server.get("/adi/enum.json")
-async def get_enum():
-    return dummy.get_enum()
-
-
-@server.get("/adi/data.json")
-async def get_data():
-    return dummy.get_data()
+@server.get("/adi/{path}")
+async def get(request: Request):
+    return dummy.process_get(request.path_params["path"], request.query_params)

schunk_egu_egk_gripper_dummy/src/dummy.py

@@ -1,5 +1,6 @@
 from threading import Thread
 import time
+from urllib.parse import parse_qs
 
 
 class Dummy(object):
@@ -19,20 +20,20 @@ def stop(self) -> None:
         self.thread.join()
         self.running = False
 
-    def _run(self):
+    def _run(self) -> None:
         while not self.done:
             time.sleep(1)
         print("Done")
 
-    def process(self, msg: str) -> bool:
-        print(msg)
-        return True
-
-    def get_info(self) -> dict:
-        return {"dataformat": 0, "numadis": 123, "webversion": 1}
-
-    def get_enum(self) -> list:
-        return []
-
-    def get_data(self) -> list:
-        return []
+    def process_get(self, path: str, query: dict[str, list[str]]) -> dict | list | None:
+        print(f"path: {path}")
+        query = parse_qs(str(query))
+        print(f"query: {query}")
+
+        if path == "info.json":
+            return {}
+        if path == "enum.json":
+            return []
+        if path == "data.json":
+            return []
+        return None

schunk_egu_egk_gripper_dummy/tests/test_dummy.py

@@ -21,23 +21,25 @@ def test_dummy_survives_repeated_starts_and_stops():
     assert not dummy.running
 
 
-def test_dummy_processes_messages():
+def test_dummy_responds_correctly_to_info_requests():
     dummy = Dummy()
-    msg = "Hello simulator!"
-    assert dummy.process(msg)
+    path = "info.json"
+    query = ""
+    expected = {}
+    assert dummy.process_get(path, query) == expected
 
 
-def test_dummy_returns_valid_info():
+def test_dummy_responds_correctly_to_enum_requests():
     dummy = Dummy()
-    info = {"dataformat": 0, "numadis": 123, "webversion": 1}
-    assert dummy.get_info() == info
+    path = "enum.json"
+    query = ""
+    expected = []
+    assert dummy.process_get(path, query) == expected
 
 
-def test_dummy_returns_valid_enums():
+def test_dummy_responds_correctly_to_data_requests():
     dummy = Dummy()
-    assert dummy.get_enum() == []
-
-
-def test_dummy_returns_valid_data():
-    dummy = Dummy()
-    assert dummy.get_data() == []
+    path = "data.json"
+    query = ""
+    expected = []
+    assert dummy.process_get(path, query) == expected