CODEWITHSUNDEEP
python
Dinero
Dinero

Reputation: 1160

Putting try and catch in the right place

I have a very simple caching service that caches files on S3. There are times when the file I am trying to cache locally does not exist on AWS S3. As such in one of my files that uses the caching service I prefer to return None if the file i am trying to cache is no found.

However I realize that i will be using the caching service in many other places and as a result I have been told by my peers that cache.cache_file() should still raise an error in this case but a simpler one like FileNotFoundError that doesn't require the caller to do if e.response["Error"]["Code"] == "404"

My Caching Code

import logging
import os
from pathlib import Path
from stat import S_IREAD, S_IRGRP, S_IROTH

from mylib import s3
from mylib.aws.clients import s3_client, s3_resource

logger = logging.getLogger(__name__)


class Cache:
    def _is_file_size_equal(self, s3_path_of_file: str, local_path: Path, file_name: str) -> bool:
        bucket, key = s3.deconstruct_s3_url(f"{s3_path_of_file}/{file_name}")
        s3_file_size = s3_resource().Object(bucket, key).content_length
        local_file_size = (local_path / file_name).stat().st_size
        return s3_file_size == local_file_size

    def cache_file(self, s3_path_of_file: str, local_path: Path, file_name: str) -> None:
        bucket, key = s3.deconstruct_s3_url(f"{s3_path_of_file}/{file_name}")
        if not (local_path / file_name).exists() or not self._is_file_size_equal(
            s3_path_of_file, local_path, file_name
        ):
            os.makedirs(local_path, exist_ok=True)
            s3_client().download_file(bucket, key, f"{local_path}/{file_name}")
            os.chmod(local_path / file_name, S_IREAD | S_IRGRP | S_IROTH)
        else:
            logger.info("Cached File is Valid!")

My Code that calls the Caching Code

def get_required_stream(environment: str, proxy_key: int) -> Optional[BinaryIO]:
    s3_overview_file_path = f"s3://{TRACK_BUCKET}/{environment}"
    overview_file = f"{some_key}.mv"
    local_path = _cache_directory(environment)
    try:
        cache.cache_file(s3_overview_file_path, local_path, overview_file)
        overview_file_cache = local_path / f"{proxy_key}.mv"
        return overview_file_cache.open("rb")
    except botocore.exceptions.ClientError as e:
        if e.response["Error"]["Code"] == "404":
            return None
        else:
            raise

Issue

Being new to Python I am a little unsure how this would work. I assume it means that my code that calls the caching service especially the except part would look something like this.

 except FileNotFoundError:
     return None

And in the caching service where i have s3_client().download_file(bucket, key, f"{local_path}/{file_name}") I would wrap it with a try and catch ?

While this question probably comes across as trivial and it is I thought I would ask it here anyway since it would be good learning opportunity for me and also understand how to write clean code. I would love suggestions on how I can achieve the desired and if my assumption is wrong?

Upvotes: 0

Views: 100

Answers (3)

Wups
Wups

Reputation: 2569

One clean way would be an additional file_exists() method in the Cache class, so every user of the cache can check before they attempt the actual caching/download. Just like using pythons filesystem/path functions.

An exception can still occur if the file is deleted/becomes unreachable between the file_exists() call and the download, but I think in this rare case, the botocore exception is just fine.

Upvotes: 1

Cyril Jouve
Cyril Jouve

Reputation: 1040

def get_required_stream(environment: str, proxy_key: int) -> Optional[BinaryIO]:
    s3_overview_file_path = f"s3://{TRACK_BUCKET}/{environment}"
    overview_file = f"{some_key}.mv"
    local_path = _cache_directory(environment)
    try:
        cache.cache_file(s3_overview_file_path, local_path, overview_file)
        overview_file_cache = local_path / f"{proxy_key}.mv"
        return overview_file_cache.open("rb")
    except botocore.exceptions.ClientError as e:
        if e.response["Error"]["Code"] == "404":
            exc = FileNotFoundError()
            exc.filename = overview_file_cache
            raise exc
        raise

# then you can use your function like this
try:
    filedesc = get_required_stream(...)
except FileNotFoundError e:
    print(f'{e.filename} not found')

Upvotes: 1

ranjith
ranjith

Reputation: 361

If you do not want calling code to catch botocore.exceptions.ClientError exception, you could wrap your cache_file method in a try except block and throw a specific exception. I would also go a step further and create a simple custom exception object that wraps botocore.exceptions.ClientError and exposes error_code and error_message from boto exception. That way, caller doesn't have to separately catch FileNotFoundError when file not found and then again botocore.exceptions.ClientError for a different type of error (say permission or some network error). They can just catch the custom exception and further inspect for more details.

try:
    //do something
except botocore.exceptions.ClientError as ex:
    raise YourCustomS3Exception(ex)   //YourCustomS3Exception needs to handle ex

Upvotes: 1

Related Questions