Repository: Netflix/consoleme
File: consoleme/lib/s3_helpers.py (Python)

Code before:
import sys
from datetime import datetime
from typing import Any, Dict, List, Optional, Union

import boto3
import pytz
import ujson as json
from asgiref.sync import sync_to_async
from botocore.exceptions import ClientError
from cloudaux import sts_conn
from cloudaux.aws.decorators import rate_limited
from cloudaux.aws.sts import boto3_cached_conn

from consoleme.config import config
from consoleme.exceptions.exceptions import MissingConfigurationValue
from consoleme.lib.plugins import get_plugin_by_name

log = config.get_logger("consoleme")
stats = get_plugin_by_name(config.get("plugins.metrics", "default_metrics"))()


async def is_object_older_than_seconds(
    bucket: Optional[str], key: str, older_than_seconds: int, s3_client=None
) -> bool:
    """
    This function checks if an S3 object is older than the specified number of seconds. if the object doesn't
    exist, this function will return True.
    """
    if not bucket:
        bucket = config.get("consoleme_s3_bucket")
    if not bucket:
        raise MissingConfigurationValue(
            "`bucket` not defined, and we can't find the default bucket in "
            "the configuration key `consoleme_s3_bucket`."
        )
    now = datetime.utcnow().replace(tzinfo=pytz.utc)
    if not s3_client:
        s3_client = boto3.client("s3")
    try:
        res = await sync_to_async(s3_client.head_object)(Bucket=bucket, Key=key)
    except ClientError as e:
        # If file is not found, we'll tell the user it's older than the specified time
        if e.response.get("Error", {}).get("Code") == "404":
            return True
        raise
    datetime_value = res["LastModified"]
    if (now - datetime_value).total_seconds() > older_than_seconds:
        return True

Review comment (suggestion):
Should this function signature changed to make `bucket` a kwarg? The call is kind of awkward with an optional positional:

```python
is_object_older_than_seconds(None, 'a/key.json', 1234)
```

On the other hand, the kwarg approach is a tad awkward when you _do_ specify a bucket:

```python
is_object_older_than_seconds('this/is/the/key.json', 1234, bucket='a-bucket')
```

Code after:
import sys
from datetime import datetime
from typing import Any, Dict, List, Optional, Union

import boto3
import pytz
import ujson as json
from asgiref.sync import sync_to_async
from botocore.exceptions import ClientError
from cloudaux import sts_conn
from cloudaux.aws.decorators import rate_limited
from cloudaux.aws.sts import boto3_cached_conn

from consoleme.config import config
from consoleme.exceptions.exceptions import MissingConfigurationValue
from consoleme.lib.plugins import get_plugin_by_name

log = config.get_logger("consoleme")
stats = get_plugin_by_name(config.get("plugins.metrics", "default_metrics"))()


async def is_object_older_than_seconds(
    key: str, older_than_seconds: int, bucket: Optional[str] = None, s3_client=None
) -> bool:
    """
    This function checks if an S3 object is older than the specified number of seconds. if the object doesn't
    exist, this function will return True.
    """
    if not bucket:
        bucket = config.get("consoleme_s3_bucket")
    if not bucket:
        raise MissingConfigurationValue(
            "`bucket` not defined, and we can't find the default bucket in "
            "the configuration key `consoleme_s3_bucket`."
        )
    now = datetime.utcnow().replace(tzinfo=pytz.utc)
    if not s3_client:
        s3_client = boto3.client("s3")
    try:
        res = await sync_to_async(s3_client.head_object)(Bucket=bucket, Key=key)
    except ClientError as e:
        # If file is not found, we'll tell the user it's older than the specified time
        if e.response.get("Error", {}).get("Code") == "404":
            return True
        raise
    datetime_value = res["LastModified"]
    if (now - datetime_value).total_seconds() > older_than_seconds:
        return True