Created
August 22, 2023 09:47
-
-
Save odinokov/e55599a26b1a488966e1dbf1ec1f7fc1 to your computer and use it in GitHub Desktop.
Lists all files in the specified S3 bucket
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import boto3 | |
from botocore.exceptions import NoCredentialsError, BotoCoreError, ClientError | |
from typing import List | |
import logging | |
# Setting up logging | |
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') | |
def list_files_from_aws( | |
region: str, path: str, aws_access_key_id: str, aws_secret_access_key: str, log_level: str = 'INFO') -> List[str]: | |
""" | |
Lists all files in the specified S3 bucket. | |
Parameters: | |
- region: AWS region where the S3 bucket is located. | |
- path: Name of the S3 bucket. | |
- aws_access_key_id: AWS access key ID for authentication. | |
- aws_secret_access_key: AWS secret access key for authentication. | |
- log_level: Logging verbosity level. | |
Returns: | |
- List containing names of the files in the bucket. | |
""" | |
logging_levels = { | |
'DEBUG': logging.DEBUG, | |
'INFO': logging.INFO, | |
'WARNING': logging.WARNING, | |
'ERROR': logging.ERROR, | |
'CRITICAL': logging.CRITICAL | |
} | |
logging.basicConfig( | |
level=logging_levels.get(log_level, logging.INFO), | |
format='%(asctime)s - %(levelname)s - %(message)s') | |
# Create a session with provided credentials | |
session = boto3.Session( | |
aws_access_key_id=aws_access_key_id, | |
aws_secret_access_key=aws_secret_access_key, | |
region_name=region | |
) | |
# Connect to the S3 service | |
s3 = session.client('s3') | |
results = [] | |
try: | |
# Using paginator for efficient retrieval | |
paginator = s3.get_paginator('list_objects_v2') | |
for page in paginator.paginate(Bucket=path): | |
for obj in page.get('Contents', []): | |
results.append(obj['Key']) | |
logging.debug(f"Retrieved file: {obj['Key']}") | |
except NoCredentialsError: | |
logging.error("Credentials not available") | |
raise | |
except (BotoCoreError, ClientError) as e: | |
logging.error(f"An error occurred: {e}") | |
raise | |
return results |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment