support for empty operation suffix

pyinfra/callback.py

@@ -66,8 +66,7 @@ class Callback:
 
     def __call__(self, analysis_input: dict):
         """data_metadata_pack: {'dossierId': ..., 'fileId': ..., 'pages': ..., 'operation': ...}"""
-        # TODO: hardcoding of "submit" is complectig callback with analysis server implementation detail
-        operation = analysis_input.get("operation", "submit")
+        operation = analysis_input.get("operation", "")
         endpoint = self.__make_endpoint(operation)
         pipeline = self.__get_pipeline(endpoint)
 

pyinfra/server/server.py

@@ -92,6 +92,10 @@ def __set_up_processing_server(operation2function: Dict[str, QueuedStreamFunctio
     def submit(operation):
         return operation2processor[operation].push(request)
 
+    @app.route("/", methods=["POST", "PATCH"])
+    def submit_default():
+        return operation2processor[""].push(request)
+
     @app.route("/<operation>", methods=["GET"])
     def pickup(operation):
         with operation2metric[operation].time():
@@ -105,7 +109,7 @@ def build_endpoint_suffixes(op: str):
 
 
 def submit_suffix(op: str):
-    return "submit" if not op else op
+    return "" if not op else op
 
 
 def pickup_suffix(op: str):

pyinfra/server/stream/rest.py

@@ -1,6 +1,7 @@
 import logging
 
 from flask import jsonify
+from funcy import drop
 
 from pyinfra.server.dispatcher.dispatcher import Nothing
 from pyinfra.server.stream.queued_stream_function import QueuedStreamFunction
@@ -16,7 +17,7 @@ class LazyRestProcessor:
 
     def push(self, request):
         self.queued_stream_function.push(request.json)
-        return jsonify(request.base_url.replace(self.submit_suffix, self.pickup_suffix))
+        return jsonify(replace_suffix(request.base_url, self.submit_suffix, self.pickup_suffix))
 
     def pop(self):
         result = self.queued_stream_function.pop() or Nothing
@@ -38,3 +39,11 @@ class LazyRestProcessor:
 
 def valid(result):
     return isinstance(result, dict) or result is Nothing
+
+
+def replace_suffix(strn, suf, repl):
+    return remove_last_n(strn, len(suf)) + repl
+
+
+def remove_last_n(strn, n):
+    return "".join(reversed(list(drop(n, reversed(strn)))))

test/config.yaml

@@ -8,19 +8,12 @@ service:
       output:
         subdir: op_outp_files
         extension: OUT.gz
-    single_inp_op:
-      input:
-        subdir: ""
-        extension: IN.gz
-      output:
-        subdir: op_outp_files
-        extension: OUT.gz
     default:
       input:
         subdir: ""
         extension: IN.gz
       output:
-        subdir: ""
+        subdir: op_outp_files
         extension: OUT.gz
 
 storage:
@@ -48,5 +41,5 @@ webserver:
 
 mock_analysis_endpoint: "http://127.0.0.1:5000"
 
-use_docker_fixture: 1
+use_docker_fixture: 0
 logging: 0

test/fixtures/input.py

@@ -4,7 +4,7 @@ from itertools import starmap, repeat
 import numpy as np
 import pytest
 from PIL import Image
-from funcy import lmap, compose, flatten, lflatten, omit, second, first, lzip
+from funcy import lmap, compose, flatten, lflatten, omit, second, first, lzip, merge
 
 from pyinfra.server.dispatcher.dispatcher import Nothing
 from pyinfra.server.normalization import normalize_item
@@ -85,9 +85,7 @@ def targets(data_message_pairs, input_data_items, operation, metadata, server_si
     try:
         response_data, response_metadata = zip(*map(unpack, flatten(starmap(op, zip(input_data_items, metadata)))))
 
-        queue_message_keys = ["id"] * (not server_side_test) + [
-            *first(queue_message_metadata).keys()
-        ]
+        queue_message_keys = ["id"] * (not server_side_test) + [*first(queue_message_metadata).keys()]
         response_metadata = lmap(partial(omit, keys=queue_message_keys), response_metadata)
         expected = lzip(response_data, response_metadata)
 
@@ -150,12 +148,14 @@ def metadata(n_items, many_to_n):
 @pytest.fixture
 def queue_message_metadata(n_items, operation_name):
     def metadata(i):
-        return {
-            "dossierId": "folder",
-            "fileId": f"file{i}",
-            "pages": [0, 2, 3],
-            "operation": operation_name,
-        }
+        return merge(
+            {
+                "dossierId": "folder",
+                "fileId": f"file{i}",
+                "pages": [0, 2, 3],
+            },
+            ({"operation": operation_name} if operation_name else {}),
+        )
 
     return lmap(metadata, range(n_items))
 

test/fixtures/server.py

@@ -52,7 +52,7 @@ def server(server_stream_function, buffer_size, operation_name):
 
 @pytest.fixture
 def operation_name(many_to_n):
-    return "multi_inp_op" if many_to_n else "single_inp_op"
+    return "multi_inp_op" if many_to_n else ""
 
 
 @pytest.fixture