From 6c82a24aa994d3c3d6bdbfee8871f6448d595c34 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pablo=20Arag=C3=B3n?= Date: Thu, 16 Jan 2025 14:28:51 +0100 Subject: [PATCH] fix lint --- gateway/api/services/result_storage.py | 24 ++++++++++++++++-------- 1 file changed, 16 insertions(+), 8 deletions(-) diff --git a/gateway/api/services/result_storage.py b/gateway/api/services/result_storage.py index 23fbdfb59..028dad187 100644 --- a/gateway/api/services/result_storage.py +++ b/gateway/api/services/result_storage.py @@ -1,11 +1,12 @@ - +""" +This module handle the access to the result store +""" import os import logging import mimetypes from typing import Optional, Tuple from wsgiref.util import FileWrapper from django.conf import settings -from django.core.files import File logger = logging.getLogger("gateway") @@ -19,12 +20,15 @@ class ResultStorage: def __init__(self, username: str): """Initialize the storage path for a given user.""" self.user_results_directory = os.path.join( - settings.MEDIA_ROOT, username, "results") + settings.MEDIA_ROOT, username, "results" + ) os.makedirs(self.user_results_directory, exist_ok=True) def __build_result_path(self, job_id: str) -> str: """Construct the full path for a result file.""" - return os.path.join(self.user_results_directory, f"{job_id}{self.RESULT_FILE_EXTENSION}") + return os.path.join( + self.user_results_directory, f"{job_id}{self.RESULT_FILE_EXTENSION}" + ) def get(self, job_id: str) -> Optional[Tuple[FileWrapper, str, int]]: """ @@ -48,8 +52,9 @@ def get(self, job_id: str) -> Optional[Tuple[FileWrapper, str, int]]: with open(result_path, "rb") as result_file: file_wrapper = FileWrapper(result_file) - file_type = mimetypes.guess_type( - result_path)[0] or "application/octet-stream" + file_type = ( + mimetypes.guess_type(result_path)[0] or "application/octet-stream" + ) file_size = os.path.getsize(result_path) return file_wrapper, file_type, file_size @@ -58,7 +63,7 @@ def save(self, job_id: str, result: str) -> None: Save the result content to a file associated with the given job ID. Args: - job_id (str): The unique identifier for the job. This will be used as the base + job_id (str): The unique identifier for the job. This will be used as the base name for the result file. result (str): The job result content to be saved in the file. """ @@ -67,4 +72,7 @@ def save(self, job_id: str, result: str) -> None: with open(result_path, "w", encoding=self.ENCODING) as result_file: result_file.write(result) logger.info( - "Result for job ID '%s' successfully saved at '%s'.", job_id, result_path) + "Result for job ID '%s' successfully saved at '%s'.", + job_id, + result_path, + )