71 lines
2.4 KiB
Python
71 lines
2.4 KiB
Python
import gzip
|
|
import json
|
|
import logging
|
|
from operator import itemgetter
|
|
|
|
import requests
|
|
|
|
from pyinfra.config import CONFIG
|
|
from pyinfra.exceptions import DataLoadingFailure, AnalysisFailure, ProcessingFailure
|
|
from pyinfra.utils.file import combine_dossier_id_and_file_id_and_extension
|
|
|
|
|
|
def make_storage_data_loader(storage, bucket_name):
|
|
def get_object_name(payload: dict) -> str:
|
|
dossier_id, file_id = itemgetter("dossierId", "fileId")(payload)
|
|
object_name = combine_dossier_id_and_file_id_and_extension(
|
|
dossier_id, file_id, CONFIG.storage.target_file_extension
|
|
)
|
|
return object_name
|
|
|
|
def download(payload):
|
|
object_name = get_object_name(payload)
|
|
logging.debug(f"Downloading {object_name}...")
|
|
data = storage.get_object(bucket_name, object_name)
|
|
logging.debug(f"Downloaded {object_name}.")
|
|
return data
|
|
|
|
def decompress(data):
|
|
return gzip.decompress(data)
|
|
|
|
def load_data(payload):
|
|
try:
|
|
return decompress(download(payload))
|
|
except Exception as err:
|
|
logging.warning(f"Loading data from storage failed for {payload}.")
|
|
raise DataLoadingFailure() from err
|
|
|
|
return load_data
|
|
|
|
|
|
def make_analyzer(analysis_endpoint):
|
|
def analyze(data):
|
|
try:
|
|
logging.debug(f"Requesting analysis from {analysis_endpoint}...")
|
|
analysis_response = requests.post(analysis_endpoint, data=data)
|
|
analysis_response.raise_for_status()
|
|
analysis_response = analysis_response.json()
|
|
logging.debug(f"Received response.")
|
|
return analysis_response
|
|
except Exception as err:
|
|
logging.warning("Exception caught when calling analysis endpoint.")
|
|
raise AnalysisFailure() from err
|
|
|
|
return analyze
|
|
|
|
|
|
def make_payload_processor(load_data, analyze_file):
|
|
def process(payload: dict):
|
|
logging.info(f"Processing {payload}...")
|
|
try:
|
|
payload = json.loads(payload)
|
|
dossier_id, file_id = itemgetter("dossierId", "fileId")(payload)
|
|
data = load_data(payload)
|
|
predictions = analyze_file(data)
|
|
return dossier_id, file_id, predictions
|
|
except (DataLoadingFailure, AnalysisFailure) as err:
|
|
logging.warning(f"Processing of {payload} failed.")
|
|
raise ProcessingFailure() from err
|
|
|
|
return process
|