Marmelator/pyinfra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 36 Wiki Activity

completed correcting / cleaning upload and download logic with regard to operations and ids. next: remove debug code

Browse Source
This commit is contained in:
Matthias Bisping 2022-07-06 16:31:50 +02:00
parent 33246d1ff9
commit 23a533e8f9
7 changed files with 66 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
pyinfra/file_descriptor_builder.py
View File

@ -38,6 +38,26 @@ class RedFileDescriptorBuilder(FileDescriptorBuilder):
def build_file_descriptor(self, queue_item_body, end="input"):
def pages():
# return [queue_item_body["id"]] if end == "output" else []
if end == "input":
print("?!?!?!?!?!")
if "id" in queue_item_body:
return [queue_item_body["id"]]
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 []
elif end == "output":
# return [queue_item_body["id"]] if "id" in queue_item_body else queue_item_body["pages"]
return [queue_item_body["id"]]
else:
raise ValueError(f"Invalid argument: {end=}") # TODO: use an enum for `end`
operation = queue_item_body.get("operation", self.default_operation_name)
file_pattern = self.operation2file_patterns[operation][end]
@ -46,7 +66,7 @@ class RedFileDescriptorBuilder(FileDescriptorBuilder):
file_descriptor = {
**project(queue_item_body, ["dossierId", "fileId", "pages"]),
"pages": [queue_item_body["id"]] if end == "output" else [],
"pages": pages(),
"extension": file_pattern["extension"],
"subdir": file_pattern["subdir"],
}

2
pyinfra/file_descriptor_manager.py
View File

@ -46,11 +46,13 @@ class FileDescriptorManager:
return self.get_object_descriptor(queue_item_body, end="input")
def get_output_object_descriptor(self, storage_upload_info):
print(1111111111111111111111111111111111111111111)
r = self.get_object_descriptor(storage_upload_info, end="output")
print(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>", r)
return r
def get_object_descriptor(self, queue_item_body, end):
print(22222222222222222222222222222222222222222, end, queue_item_body)
# TODO: this is complected with the Storage class API
# FIXME: bad coupling
return {

3
pyinfra/visitor/downloader.py
View File

@ -51,7 +51,10 @@ class Downloader:
yield from self.storage.get_all_object_names(self.bucket_name, prefix=prefix)
def get_pattern_filter(self, queue_item_body):
print("<<<<<<<<<<<<<<<<<<<<<<<<<<<", 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]+'...')}")
matches_pattern = flift(file_pattern)
return matches_pattern

17
test/config.yaml
View File

@ -1,17 +1,11 @@
service:
response_formatter: identity
operations:
# multi_inp_op:
# input:
# subdir: op_inp_files
# extension: IN.gz
# output:
# subdir: op_outp_files
# extension: OUT.gz
upper:
input:
subdir: ""
extension: up_in.gz
multi: False
output:
subdir: ""
extension: up_out.gz
@ -19,6 +13,7 @@ service:
input:
subdir: ""
extension: extr_in.gz
multi: False
output:
subdir: "extractions"
extension: gz
@ -26,6 +21,7 @@ service:
input:
subdir: ""
extension: rot_in.gz
multi: False
output:
subdir: ""
extension: rot_out.gz
@ -33,6 +29,7 @@ service:
input:
subdir: ""
extension: cls_in.gz
multi: True
output:
subdir: ""
extension: cls_out.gz
@ -40,15 +37,15 @@ service:
input:
subdir: ""
extension: pgs_in.gz
multi: False
output:
subdir: "pages"
extension: pgs_out.gz
default:
input:
subdir: ""
extension: IN.gz
multi: False
output:
subdir: ""
extension: OUT.gz
@ -79,4 +76,4 @@ webserver:
mock_analysis_endpoint: "http://127.0.0.1:5000"
use_docker_fixture: 0
logging: 1
logging: 0

4
test/conftest.py
View File

@ -89,8 +89,8 @@ def mock_make_load_data():
return load_data
def pytest_make_parametrize_id(config, val, argname):
return f"\n\t{argname}={val}\n"
# def pytest_make_parametrize_id(config, val, argname):
# return f"\n\t{argname}={val}\n"
@pytest.fixture

36
test/fixtures/server.py vendored
View File

@ -102,7 +102,7 @@ def server_side_test(request):
@pytest.fixture
def core_operation(item_type, one_to_many, analysis_task):
def core_operation(many_to_n, item_type, one_to_many, analysis_task):
def duplicate(string: bytes, metadata):
for _ in range(2):
yield upper(string, metadata), metadata
@ -117,7 +117,9 @@ def core_operation(item_type, one_to_many, analysis_task):
def extract(string: bytes, metadata):
print("extract>>>>>>>>>>>", metadata)
print(string.decode())
for i, c in project(dict(enumerate(string.decode())), metadata.get("pages", range(len(string.decode())))).items():
for i, c in project(
dict(enumerate(string.decode())), metadata.get("pages", range(len(string.decode())))
).items():
metadata["id"] = i
print(i)
yield c.encode(), metadata
@ -136,18 +138,34 @@ def core_operation(item_type, one_to_many, analysis_task):
yield f"page_{i}".encode(), metadata
params2op = {
False: {
"string": {False: upper},
"image": {False: rotate, True: classify},
},
True: {
"string": {False: extract},
"pdf": {False: stream_pages},
True: {},
False: {
"image": {True: classify},
},
},
False: {
True: {
"string": {False: extract},
"pdf": {False: stream_pages},
},
False: {
"string": {False: upper},
"image": {False: rotate},
},
},
# False: {
# "string": {False: upper},
# "image": {False: rotate, True: classify},
# },
# True: {
# "string": {False: extract},
# "pdf": {False: stream_pages},
# },
}
try:
return params2op[one_to_many][item_type][analysis_task]
return params2op[many_to_n][one_to_many][item_type][analysis_task]
except KeyError:
msg = f"No operation defined for [{one_to_many=}, {item_type=}, {analysis_task=}]."
pytest.skip(msg)

6
test/integration_tests/serve_test.py
View File

@ -78,7 +78,7 @@ from test.config import CONFIG as TEST_CONFIG
@pytest.mark.parametrize(
"many_to_n",
[
# True,
True,
False,
],
)
@ -157,8 +157,9 @@ def upload_data_to_folder_in_storage_and_publish_single_request_to_queue(
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["object_name"] = build_filepath(object_descriptor, page)
print(f"{object_descriptor=}")
object_descriptor["object_name"] = build_filepath(object_descriptor, page)
# input()
storage.put_object(**object_descriptor, data=compress(data))
@ -170,6 +171,7 @@ def build_filepath(object_descriptor, page):
parts = object_name.split("/")
path = "/".join(parts)
path = re.sub(r"id:\d", f"id:{page}", path)
print(path)
return path