6
\$\begingroup\$

Background

I have a code that basically caches file. The way it works is it will first check if the file exists on the local drive and if it does if there is a flag passed called validate_file_size then it will do another check to see if the size of the file on local drive equals the size of the file on S3. if they equal we assume the local file is the most recent version and if they don't then we need to download the latest version.

CODE

  def cache_file(self, s3_path_of_file: str, local_path: Path, file_name: str, validate_file_size: bool = True) -> None:
        bucket, key = s3.deconstruct_s3_url(f"{s3_path_of_file}/{file_name}")
        is_file_size_valid = True
        if not (local_path / file_name).exists():
            os.makedirs(local_path, exist_ok=True)
            try:
                s3_client().download_file(bucket, key, f"{local_path}/{file_name}")
                os.chmod(local_path / file_name, S_IREAD | S_IRGRP | S_IROTH)
            except botocore.exceptions.ClientError as e:
                if e.response["Error"]["Code"] == "404":
                    raise FileNotFoundError
        elif validate_file_size:
            is_file_size_valid = self._is_file_size_equal(
                s3_path_of_file, local_path, file_name
            )
            if not is_file_size_valid:
                try:
                    s3_client().download_file(bucket, key, f"{local_path}/{file_name}")
                    os.chmod(local_path / file_name, S_IREAD | S_IRGRP | S_IROTH)
                except botocore.exceptions.ClientError as e:
                    if e.response["Error"]["Code"] == "404":
                        raise FileNotFoundError
            else:
                logger.info("Cached File is Valid!")

ISSUE

I am wondering if there is a better way to organize my if else statements so my code is cleaner and less redundant. I am not particularly happy about this block in particular that is repeated twice.

 try:
     s3_client().download_file(bucket, key, f"{local_path}/{file_name}")
     os.chmod(local_path / file_name, S_IREAD | S_IRGRP | S_IROTH)
 except botocore.exceptions.ClientError as e:
     if e.response["Error"]["Code"] == "404":
         raise FileNotFoundError

One obvious choice is to take it out as a separate function but I am almost wondering if i can avoid that and organize my if else statement in a way where i don't really repeat this try except block twice.

\$\endgroup\$
1

5 Answers 5

7
\$\begingroup\$

I'm not a Python programmer, but the way to not repeat the code without creating a function is to do the download checking first have flags set by that checking one for creating a new file path and one for doing the download. Figure out everything that needs to be done first and then do the download and path creation as necessary.

pseudo code:

if not local path or file:
    create_local_path_flag = True

if file size not valid:
    file_out_of_date_flag = True

if create_local_path_flag or file_out_of_date_flag:
    if create_local_path_flag:
        create_local_path()
    download_file()
\$\endgroup\$
3
  • 2
    \$\begingroup\$ IMO this is not a very pythonic approach, setting multiple boolean flags to avoid function calls doesn't seem like the way I would go. I think OP is on the right track with extracting a function. \$\endgroup\$
    – David258
    Aug 31, 2020 at 10:50
  • 2
    \$\begingroup\$ You missed the last paragraph, the OP doesn't want to extract the function, or that was the way the last paragraph was originally phrased. \$\endgroup\$
    – pacmaninbw
    Aug 31, 2020 at 12:06
  • 4
    \$\begingroup\$ -fair enough, I think we had different readings of the question. I read it as "is using a function a good way or is there better?", to which my answer is "using a function is the better way". \$\endgroup\$
    – David258
    Aug 31, 2020 at 12:21
6
\$\begingroup\$

Perhaps this does what you need:

exists_locally = (local_path / file_name).exists()
sizes_agree = (
    validate_file_size is False or
    (
        exists_locally and
        self._is_file_size_equal(s3_path_of_file, local_path, file_name)
    )
)
if exists_locally and sizes_agree:
    logger.info('All is well')
else:
    os.makedirs(local_path, exist_ok=True)
    self.download_file(...)
    logger.info('Blah blah')

The code could be further simplified if self._is_file_size_equal() will happily return False when the local file is absent.

\$\endgroup\$
4
  • \$\begingroup\$ You have unbalanced parentheses in the exists_locally assignment. You're also missing concatenation of the /, although it would be better to use os.path.join(). \$\endgroup\$
    – Barmar
    Aug 31, 2020 at 15:48
  • \$\begingroup\$ @Barmar "You're also missing concatenation of the /" isn't that what local_path / file_name is doing? \$\endgroup\$
    – Peilonrayz
    Aug 31, 2020 at 15:50
  • \$\begingroup\$ I didn't notice that local_path was a class that redefines the / operator. \$\endgroup\$
    – Barmar
    Aug 31, 2020 at 16:00
  • \$\begingroup\$ @Barmar Thanks, fixed. \$\endgroup\$
    – FMc
    Aug 31, 2020 at 16:08
4
\$\begingroup\$

You're natural response to separate it out into a function is the right one.

Even if you do re-organise your code such that you only need this block once and its not repeated, splitting it out into it's own function is still a good idea as it is much easier to test standalone, and would be cleaner to refactor if you want to use a different storage provider/access library in future.

Further things to add:

  • If you make the function standalone, then you could pull in the bucket, key = s3... to the new download_s3_file(s3_path, local_filepath) function
  • You should add a raise at the end of the exception if you want to catch errors other than 404 (currently e.g. 500 would pass silently)
  • You should add the paths to the logs/raises to help yourself later
  • You only log on "Cache is valid", I would either log on everything ("Downloading file", "Validating cache") or nothing; everything is probably better!
  • You can use Path.mkdir() rather than os.makedir()
  • You don't use the is_file_size_valid variable after setting it
  • As a matter of personal taste, I would combine the local_path and file_name as early as possible (probably before this function!)
  • As a matter of personal taste, I would return the local_filepath from this function
def cache_file(self, s3_path_of_file: str, local_path: Path, file_name: str, validate_file_size: bool = True) -> Path:
    s3_path = f"{s3_path_of_file}/{file_name}"
    local_filepath = f"{local_path}/{file_name}"
    
    if not local_filepath.exists():
        logger.info(f"File does not exist locally, downloading from s3: {s3_path} -> {local_filepath}")
        local_filepath.parent.mkdir(exist_ok=True, parents=True)
        download_s3_file(s3_path, local_filepath)   
    elif validate_file_size and not self._is_file_size_equal(s3_path, local_filepath):
        logger.info(f"Cache file is invalid, re-downloading: {local_filepath}")
        download_s3_file(s3_path, local_filepath)
    else: 
        logger.info(f"Cache file exists locally: {local_filepath}")

    return local_filepath

            
def download_s3_file(s3_path, local_path)
    bucket, key = s3.deconstruct_s3_url(s3_path)
    try:
        s3_client().download_file(bucket, key, local_path)
        os.chmod(local_path / file_name, S_IREAD | S_IRGRP | S_IROTH)
    except botocore.exceptions.ClientError as e:
        if e.response["Error"]["Code"] == "404":
            raise FileNotFoundError("File not found on S3: {s3_path}")
        raise
```
\$\endgroup\$
4
\$\begingroup\$

Why avoid extracting it to a separate function? I'd argue that extracting it would make the cache_file function easier to read (with good function naming).

Similarly, checking whether or not the file should be downloaded can be extracted to a function, say, is_latest_version_cached. The cache_file function would then look something like this:

  def cache_file(self, s3_path_of_file: str, local_path: Path, file_name: str, validate_file_size: bool = True) -> None:
        if not _is_latest_version_cached(): # input arguments not shown
            try:
                s3_client().download_file(bucket, key, f"{local_path}/{file_name}")
                os.chmod(local_path / file_name, S_IREAD | S_IRGRP | S_IROTH)
            except botocore.exceptions.ClientError as e:
                if e.response["Error"]["Code"] == "404":
                    raise FileNotFoundError
        else:
            logger.info("Cached File is Valid!")

This removes the repetition.

In addition to that, I'd extract a function that does the actual caching. The result is then:

  def cache_file(self, s3_path_of_file: str, local_path: Path, file_name: str, validate_file_size: bool = True) -> None:
        if not _is_latest_version_cached(): # input arguments not shown
            _cache_file() # input arguments not shown
        else:
            logger.info("Cached File is Valid!")

That said, the code does become a bit less neat when you add the required inputs for _is_lastest_version_cached and _cache_file that I omitted in the snippets above.

It's been a while since I programmed in python, so I'm not sure how pythonic it is to use private methods here and for cache_file and _cache_file to basically have the same name.

Perhaps it's worth considering renaming the original method to something that indicates that caching is only done if necessary, e.g. ensure_file_is_cached, if this is valuable info to the caller.

\$\endgroup\$
0
\$\begingroup\$

Invert the logic of the if statements and return early if the file doesn't need to be downloaded. makedirs() is probably much faster than downloading the file, so calling it every time a file is downloaded won't affect run time.

def cache_file(self, s3_path_of_file: str, 
                     local_path: Path,
                     file_name: str,
                     validate_file_size: bool = True) -> None:

    if (    (local_path / file_name).exists()
        and (   validate_file_size == False 
             or self._is_file_size_equal(s3_path_of_file, 
                                             local_path, file_name)
            )
       ):
        return

    os.makedirs(local_path, exist_ok=True)
    bucket, key = s3.deconstruct_s3_url(f"{s3_path_of_file}/{file_name}")

    try:
        s3_client().download_file(bucket, key, f"{local_path}/{file_name}")
        os.chmod(local_path / file_name, S_IREAD | S_IRGRP | S_IROTH)

    except botocore.exceptions.ClientError as e:
        if e.response["Error"]["Code"] == "404":
            raise FileNotFoundError
        else:
            logger.info("Cached File is Valid!")
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.