removed debug prints
This commit is contained in:
parent
23a533e8f9
commit
4287f6d987
@ -39,21 +39,12 @@ class RedFileDescriptorBuilder(FileDescriptorBuilder):
|
|||||||
def build_file_descriptor(self, queue_item_body, end="input"):
|
def build_file_descriptor(self, queue_item_body, end="input"):
|
||||||
|
|
||||||
def pages():
|
def pages():
|
||||||
# return [queue_item_body["id"]] if end == "output" else []
|
|
||||||
if end == "input":
|
if end == "input":
|
||||||
print("?!?!?!?!?!")
|
|
||||||
if "id" in queue_item_body:
|
if "id" in queue_item_body:
|
||||||
return [queue_item_body["id"]]
|
return [queue_item_body["id"]]
|
||||||
else:
|
else:
|
||||||
# return queue_item_body["pages"]
|
|
||||||
# return [] if file_pattern["subdir"] else queue_item_body["pages"]
|
|
||||||
# return [] if len(queue_item_body["pages"]) > 1 else queue_item_body["pages"]
|
|
||||||
# return [] # single input
|
|
||||||
# return queue_item_body["pages"] # multi input
|
|
||||||
print(file_pattern)
|
|
||||||
return queue_item_body["pages"] if file_pattern["multi"] else []
|
return queue_item_body["pages"] if file_pattern["multi"] else []
|
||||||
elif end == "output":
|
elif end == "output":
|
||||||
# return [queue_item_body["id"]] if "id" in queue_item_body else queue_item_body["pages"]
|
|
||||||
return [queue_item_body["id"]]
|
return [queue_item_body["id"]]
|
||||||
else:
|
else:
|
||||||
raise ValueError(f"Invalid argument: {end=}") # TODO: use an enum for `end`
|
raise ValueError(f"Invalid argument: {end=}") # TODO: use an enum for `end`
|
||||||
@ -62,8 +53,6 @@ class RedFileDescriptorBuilder(FileDescriptorBuilder):
|
|||||||
|
|
||||||
file_pattern = self.operation2file_patterns[operation][end]
|
file_pattern = self.operation2file_patterns[operation][end]
|
||||||
|
|
||||||
print(f"{end=}")
|
|
||||||
|
|
||||||
file_descriptor = {
|
file_descriptor = {
|
||||||
**project(queue_item_body, ["dossierId", "fileId", "pages"]),
|
**project(queue_item_body, ["dossierId", "fileId", "pages"]),
|
||||||
"pages": pages(),
|
"pages": pages(),
|
||||||
@ -91,14 +80,11 @@ class RedFileDescriptorBuilder(FileDescriptorBuilder):
|
|||||||
def __build_page_regex(pages, subdir):
|
def __build_page_regex(pages, subdir):
|
||||||
|
|
||||||
n_pages = len(pages)
|
n_pages = len(pages)
|
||||||
print("pages:", pages)
|
|
||||||
|
|
||||||
if n_pages > 1:
|
if n_pages > 1:
|
||||||
page_re = "id:(" + "|".join(map(str, pages)) + ")."
|
page_re = "id:(" + "|".join(map(str, pages)) + ")."
|
||||||
elif n_pages == 1:
|
elif n_pages == 1:
|
||||||
page_re = f"id:{pages[0]}."
|
page_re = f"id:{pages[0]}."
|
||||||
else: # no pages specified -> either all pages or no pages, depending on whether a subdir is specified
|
else: # no pages specified -> either all pages or no pages, depending on whether a subdir is specified
|
||||||
print("subdir", subdir)
|
|
||||||
page_re = r"id:\d+." if subdir else ""
|
page_re = r"id:\d+." if subdir else ""
|
||||||
|
|
||||||
return page_re
|
return page_re
|
||||||
|
|||||||
@ -46,13 +46,9 @@ class FileDescriptorManager:
|
|||||||
return self.get_object_descriptor(queue_item_body, end="input")
|
return self.get_object_descriptor(queue_item_body, end="input")
|
||||||
|
|
||||||
def get_output_object_descriptor(self, storage_upload_info):
|
def get_output_object_descriptor(self, storage_upload_info):
|
||||||
print(1111111111111111111111111111111111111111111)
|
return self.get_object_descriptor(storage_upload_info, end="output")
|
||||||
r = self.get_object_descriptor(storage_upload_info, end="output")
|
|
||||||
print(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>", r)
|
|
||||||
return r
|
|
||||||
|
|
||||||
def get_object_descriptor(self, queue_item_body, end):
|
def get_object_descriptor(self, queue_item_body, end):
|
||||||
print(22222222222222222222222222222222222222222, end, queue_item_body)
|
|
||||||
# TODO: this is complected with the Storage class API
|
# TODO: this is complected with the Storage class API
|
||||||
# FIXME: bad coupling
|
# FIXME: bad coupling
|
||||||
return {
|
return {
|
||||||
|
|||||||
@ -27,7 +27,6 @@ class S3StorageAdapter(StorageAdapter):
|
|||||||
logger.debug(f"Uploading '{object_name}'...")
|
logger.debug(f"Uploading '{object_name}'...")
|
||||||
data = io.BytesIO(data)
|
data = io.BytesIO(data)
|
||||||
self.__client.put_object(bucket_name, object_name, data, length=data.getbuffer().nbytes)
|
self.__client.put_object(bucket_name, object_name, data, length=data.getbuffer().nbytes)
|
||||||
# input("uploaded")
|
|
||||||
|
|
||||||
def get_object(self, bucket_name, object_name):
|
def get_object(self, bucket_name, object_name):
|
||||||
logger.debug(f"Downloading '{object_name}'...")
|
logger.debug(f"Downloading '{object_name}'...")
|
||||||
|
|||||||
@ -51,10 +51,7 @@ class Downloader:
|
|||||||
yield from self.storage.get_all_object_names(self.bucket_name, prefix=prefix)
|
yield from self.storage.get_all_object_names(self.bucket_name, prefix=prefix)
|
||||||
|
|
||||||
def get_pattern_filter(self, queue_item_body):
|
def get_pattern_filter(self, queue_item_body):
|
||||||
print("<<<<<<<<<<<<<<<<<<<<<<<<<<<", queue_item_body)
|
|
||||||
file_pattern = self.file_descriptor_manager.build_input_matcher(queue_item_body)
|
file_pattern = self.file_descriptor_manager.build_input_matcher(queue_item_body)
|
||||||
print(f"{file_pattern=}")
|
|
||||||
# input()
|
|
||||||
logger.debug(f"Filtering pattern: {file_pattern if len(file_pattern) <= 120 else (file_pattern[:120]+'...')}")
|
logger.debug(f"Filtering pattern: {file_pattern if len(file_pattern) <= 120 else (file_pattern[:120]+'...')}")
|
||||||
matches_pattern = flift(file_pattern)
|
matches_pattern = flift(file_pattern)
|
||||||
return matches_pattern
|
return matches_pattern
|
||||||
|
|||||||
9
test/fixtures/input.py
vendored
9
test/fixtures/input.py
vendored
@ -6,7 +6,6 @@ import pytest
|
|||||||
from PIL import Image
|
from PIL import Image
|
||||||
from funcy import lmap, compose, flatten, lflatten, omit, second, first, lzip, merge
|
from funcy import lmap, compose, flatten, lflatten, omit, second, first, lzip, merge
|
||||||
|
|
||||||
from pyinfra.server.debugging import inspect
|
|
||||||
from pyinfra.server.dispatcher.dispatcher import Nothing
|
from pyinfra.server.dispatcher.dispatcher import Nothing
|
||||||
from pyinfra.server.normalization import normalize_item
|
from pyinfra.server.normalization import normalize_item
|
||||||
from pyinfra.server.packing import pack, unpack
|
from pyinfra.server.packing import pack, unpack
|
||||||
@ -81,7 +80,7 @@ def targets(data_message_pairs, input_data_items, operation, metadata, server_si
|
|||||||
if operation is Nothing:
|
if operation is Nothing:
|
||||||
return Nothing
|
return Nothing
|
||||||
|
|
||||||
op = compose(lift(star(pack)), normalize_item, inspect("A"), operation)
|
op = compose(lift(star(pack)), normalize_item, operation)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
response_data, response_metadata = zip(*map(unpack, flatten(starmap(op, zip(input_data_items, metadata)))))
|
response_data, response_metadata = zip(*map(unpack, flatten(starmap(op, zip(input_data_items, metadata)))))
|
||||||
@ -148,10 +147,8 @@ def metadata(n_items, many_to_n):
|
|||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def queue_message_metadata(n_items, operation_name):
|
def queue_message_metadata(n_items, operation_name):
|
||||||
print()
|
|
||||||
print("n_items:", n_items)
|
|
||||||
def metadata(i):
|
def metadata(i):
|
||||||
r = merge(
|
return merge(
|
||||||
{
|
{
|
||||||
"dossierId": "dossier_id",
|
"dossierId": "dossier_id",
|
||||||
"fileId": f"file_id_{i}",
|
"fileId": f"file_id_{i}",
|
||||||
@ -159,8 +156,6 @@ def queue_message_metadata(n_items, operation_name):
|
|||||||
({"operation": operation_name} if operation_name else {}),
|
({"operation": operation_name} if operation_name else {}),
|
||||||
({"pages": [0, 2, 3]} if n_items > 1 else {}),
|
({"pages": [0, 2, 3]} if n_items > 1 else {}),
|
||||||
)
|
)
|
||||||
print("metadata:", r)
|
|
||||||
return r
|
|
||||||
|
|
||||||
return lmap(metadata, range(n_items))
|
return lmap(metadata, range(n_items))
|
||||||
|
|
||||||
|
|||||||
7
test/fixtures/server.py
vendored
7
test/fixtures/server.py
vendored
@ -54,7 +54,6 @@ def server(server_stream_function, buffer_size, operation_name):
|
|||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def operation_name(core_operation):
|
def operation_name(core_operation):
|
||||||
operation_name = core_operation.__name__
|
operation_name = core_operation.__name__
|
||||||
print("OPERATION NAME", operation_name)
|
|
||||||
return operation_name
|
return operation_name
|
||||||
|
|
||||||
|
|
||||||
@ -108,20 +107,14 @@ def core_operation(many_to_n, item_type, one_to_many, analysis_task):
|
|||||||
yield upper(string, metadata), metadata
|
yield upper(string, metadata), metadata
|
||||||
|
|
||||||
def upper(string: bytes, metadata):
|
def upper(string: bytes, metadata):
|
||||||
print("upper>>>>>>>>>>>", metadata)
|
|
||||||
print("???????????????????????????????", string)
|
|
||||||
r = string.decode().upper().encode(), metadata
|
r = string.decode().upper().encode(), metadata
|
||||||
print("!!!!!!!!!!!!!!!!!!!!!", r)
|
|
||||||
return r
|
return r
|
||||||
|
|
||||||
def extract(string: bytes, metadata):
|
def extract(string: bytes, metadata):
|
||||||
print("extract>>>>>>>>>>>", metadata)
|
|
||||||
print(string.decode())
|
|
||||||
for i, c in project(
|
for i, c in project(
|
||||||
dict(enumerate(string.decode())), metadata.get("pages", range(len(string.decode())))
|
dict(enumerate(string.decode())), metadata.get("pages", range(len(string.decode())))
|
||||||
).items():
|
).items():
|
||||||
metadata["id"] = i
|
metadata["id"] = i
|
||||||
print(i)
|
|
||||||
yield c.encode(), metadata
|
yield c.encode(), metadata
|
||||||
|
|
||||||
def rotate(im: bytes, metadata):
|
def rotate(im: bytes, metadata):
|
||||||
|
|||||||
@ -82,11 +82,9 @@ from test.config import CONFIG as TEST_CONFIG
|
|||||||
False,
|
False,
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
def test_serving(operation_name, server_process, bucket_name, components, targets, data_message_pairs, n_items, many_to_n):
|
def test_serving(
|
||||||
|
operation_name, server_process, bucket_name, components, targets, data_message_pairs, n_items, many_to_n
|
||||||
print(f"{operation_name=}")
|
):
|
||||||
print()
|
|
||||||
print()
|
|
||||||
|
|
||||||
storage, queue_manager, consumer, file_descriptor_manager = components
|
storage, queue_manager, consumer, file_descriptor_manager = components
|
||||||
|
|
||||||
@ -112,12 +110,6 @@ def test_serving(operation_name, server_process, bucket_name, components, target
|
|||||||
|
|
||||||
# TODO: correctness of target should be validated as well, since production has become non-trivial
|
# TODO: correctness of target should be validated as well, since production has become non-trivial
|
||||||
assert sorted(outputs) == sorted(targets)
|
assert sorted(outputs) == sorted(targets)
|
||||||
print(sorted(targets))
|
|
||||||
print(sorted(outputs))
|
|
||||||
|
|
||||||
# sleep(2)
|
|
||||||
# print("---")
|
|
||||||
# input()
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
@ -157,9 +149,7 @@ def upload_data_to_folder_in_storage_and_publish_single_request_to_queue(
|
|||||||
|
|
||||||
for data, page in zip(map(first, data_message_pairs), pages):
|
for data, page in zip(map(first, data_message_pairs), pages):
|
||||||
object_descriptor = file_descriptor_manager.get_input_object_descriptor({**ref_message, "id": page})
|
object_descriptor = file_descriptor_manager.get_input_object_descriptor({**ref_message, "id": page})
|
||||||
print(f"{object_descriptor=}")
|
|
||||||
object_descriptor["object_name"] = build_filepath(object_descriptor, page)
|
object_descriptor["object_name"] = build_filepath(object_descriptor, page)
|
||||||
# input()
|
|
||||||
|
|
||||||
storage.put_object(**object_descriptor, data=compress(data))
|
storage.put_object(**object_descriptor, data=compress(data))
|
||||||
|
|
||||||
@ -171,7 +161,6 @@ def build_filepath(object_descriptor, page):
|
|||||||
parts = object_name.split("/")
|
parts = object_name.split("/")
|
||||||
path = "/".join(parts)
|
path = "/".join(parts)
|
||||||
path = re.sub(r"id:\d", f"id:{page}", path)
|
path = re.sub(r"id:\d", f"id:{page}", path)
|
||||||
print(path)
|
|
||||||
|
|
||||||
return path
|
return path
|
||||||
|
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user