Reference
S3 Bucket helper utils. Async list objects by folders
The file is mandatory for build system to find the package.
Classes
Attributes
instance-attribute
semaphore = Semaphore(parallelism)
Functions
async
iter(prefix: str = '/', *, max_level: Optional[int] = None, max_folders: Optional[int] = None, delimiter: str = '/') -> AsyncIterator[list[dict[str, Any]]]
Generator that yields objects in the bucket with the given prefix.
Yield objects by partial chunks (list of AWS S3 object dicts) as they are collected from AWS asynchronously.
max_level: The maximum folders depth to traverse in separate requests. If None, traverse all levels. max_folders: The maximum number of folders to load in separate requests. If None, requests all folders. Otherwise, the folders are grouped by prefixes before loading in separate requests. Try to group in the given number of folders if possible. delimiter: The delimiter for "folders".
async
list(prefix: str = '/', *, max_level: Optional[int] = None, max_folders: Optional[int] = None, delimiter: str = '/') -> list[dict[str, Any]]
List all objects in the bucket with the given prefix.
max_level: The maximum folders depth to traverse in separate requests.
If None, traverse all levels.
max_folders: The maximum number of folders to load in separate requests.
If None, requests all folders.
Otherwise, the folders are grouped by prefixes before loading in separate requests.
Try to group to the given max_folders if possible.
delimiter: The delimiter for "folders".
Modules
Functions
find_longest_common_prefix(words: list[str]) -> str
Finds the longest common prefix among a list of words.
group_by_prefix(words: list[str], desired_group_count: int) -> list[str]
Groups words by prefixes to create a desired number of word groups.
Try to create the desired number of groups if possible.
async-s3.
Attributes
module-attribute
PROGRESS_REFRESH_INTERVAL = 0.5
module-attribute
S3PROTO = 's3://'
Classes
dataclass
Configuration for S3 object listing operations.
property
bucket: str
Extract bucket name from S3 URL.
class-attribute
instance-attribute
delimiter: str = '/'
property
key: str
Extract key/prefix from S3 URL.
class-attribute
instance-attribute
max_folders: Optional[int] = None
class-attribute
instance-attribute
max_level: Optional[int] = None
class-attribute
instance-attribute
parallelism: int = 100
class-attribute
instance-attribute
repeat: int = 1
instance-attribute
s3_url: str
Base class for S3 listing commands.
instance-attribute
config = ListingConfig(s3_url=s3_url, max_level=max_level, max_folders=max_folders, repeat=repeat, parallelism=parallelism, delimiter=delimiter)
validate_and_execute() -> Iterable[dict[str, Any]]
Validate S3 URL and execute listing.
Functions
as3() -> None
Async S3.
du(s3_url: str, max_level: Optional[int], max_folders: Optional[int], repeat: int, parallelism: int, delimiter: str) -> None
Show count and size for objects in an S3 bucket.
Example: as3 du s3://bucket/key
error(message: str) -> None
Print an error message and exit.
human_readable_size(size: float, decimal_places: int = 2) -> str
Convert bytes to a human-readable format.
list_objects(config: ListingConfig) -> Iterable[dict[str, Any]]
List objects in an S3 bucket.
async
list_objects_async(config: ListingConfig) -> Iterable[dict[str, Any]]
List objects in an S3 bucket.
list_objects_options(func: Callable[[Any], None]) -> Callable[[Any], None]
Add common options to commands using list_objects.
async
list_objects_with_progress(s3_list: S3BucketObjects, config: ListingConfig) -> tuple[Iterable[dict[str, Any]], float]
List objects in an S3 bucket with a progress bar.
Returns: (The objects, the elapsed time)
ls(s3_url: str, max_level: Optional[int], max_folders: Optional[int], repeat: int, parallelism: int, delimiter: str) -> None
List objects in an S3 bucket.
Example: as3 ls s3://bucket/key
print_attempt_info(attempt: int, duration: float) -> None
Print the elapsed time for an attempt.
print_average_time(total_time: float, repeat: int) -> None
Print the average elapsed time.
print_start_info(config: ListingConfig) -> None
Print the command parameters.
print_summary(objects: Iterable[dict[str, Any]]) -> None
Print a summary of the objects.
split_s3_url(s3_url: str) -> tuple[str, str]
Split an S3 URL into bucket and key.
validate_delimiter(ctx: Context, param: Parameter, value: str) -> str
Validate the Delimiter option.
Attributes
module-attribute
DEFAULT_PARALLELISM = 100
Classes
instance-attribute
semaphore = Semaphore(parallelism)
async
iter(prefix: str = '/', *, max_level: Optional[int] = None, max_folders: Optional[int] = None, delimiter: str = '/') -> AsyncIterator[list[dict[str, Any]]]
Generator that yields objects in the bucket with the given prefix.
Yield objects by partial chunks (list of AWS S3 object dicts) as they are collected from AWS asynchronously.
max_level: The maximum folders depth to traverse in separate requests. If None, traverse all levels. max_folders: The maximum number of folders to load in separate requests. If None, requests all folders. Otherwise, the folders are grouped by prefixes before loading in separate requests. Try to group in the given number of folders if possible. delimiter: The delimiter for "folders".
async
list(prefix: str = '/', *, max_level: Optional[int] = None, max_folders: Optional[int] = None, delimiter: str = '/') -> list[dict[str, Any]]
List all objects in the bucket with the given prefix.
max_level: The maximum folders depth to traverse in separate requests.
If None, traverse all levels.
max_folders: The maximum number of folders to load in separate requests.
If None, requests all folders.
Otherwise, the folders are grouped by prefixes before loading in separate requests.
Try to group to the given max_folders if possible.
delimiter: The delimiter for "folders".
Functions
cached
create_session() -> AioSession
Create a session object.
get_s3_client() -> AioBaseClient
Get S3 client.