Skip to content

Secrets Manager

Botree S3 utilities.

Classes

botree.s3.Bucket(session: Session, name: str, client_kwargs: dict = dict(), resource_kwargs: dict = dict())

AWS S3 Bucket level transactions.

Source code in botree/s3.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
def __init__(
    self,
    session: Session,
    name: str,
    client_kwargs: dict = dict(),
    resource_kwargs: dict = dict(),
):
    self.name = name
    self.session = session
    self.client = self.session.client(service_name="s3", **client_kwargs)
    self.resource = self.session.resource("s3", **resource_kwargs).Bucket(
        name=self.name
    )

Functions

copy(source: Path, target: Path, source_bucket: Optional[str] = None, *args: Optional[str], **kwargs: Optional[str])

Copy an object stored in the bucket or from another bucket.

If source_bucket is not specified, the object is copied from the current bucket.

if target_key has no suffix, it assumes target_key is a folder and source_key file name is used.

Parameters:

  • source (Path) –

    source file path

  • target (Path) –

    target file path. If target_key has no suffix, it assumes target_key is a folder.

  • source_bucket (Optional[str]) –

    bucket from wich source file will be copied from, by default None

Source code in botree/s3.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def copy(
    self,
    source: Path,
    target: Path,
    source_bucket: Optional[str] = None,
    *args,
    **kwargs,
):
    """
    Copy an object stored in the bucket or from another bucket.

    If source_bucket is not specified, the object is copied from the current bucket.

    if target_key has no suffix, it assumes target_key is a folder and source_key
    file name is used.

    Parameters
    ----------
    source : Path
        source file path
    target : Path
        target file path. If target_key has no suffix, it assumes target_key
        is a folder.
    source_bucket : Optional[str], optional
        bucket from wich source file will be copied from, by default None
    """
    copy_source = {"Bucket": source_bucket, "Key": str(source)}

    if not source_bucket:
        copy_source.update({"Bucket": self.name})

    if not target.suffix:
        target = target / source.name

    self.client.copy(copy_source, self.name, str(target), **kwargs)  # type: ignore
delete(target: Path, **kwargs: Path)

Delete a file from S3.

Parameters:

  • target (Path) –

    remote file path.

Source code in botree/s3.py 
165
166
167
168
169
170
171
172
173
174
def delete(self, target: Path, **kwargs):
    """
    Delete a file from S3.

    Parameters
    ----------
    target : Path
        remote file path.
    """
    self.client.delete_object(Bucket=self.name, Key=str(target), **kwargs)
download(source: Path, target: Path, **kwargs: Path)

Download a file from S3.

Parameters:

  • source (Path) –

    remote file path.

  • target (Path) –

    local file path.

Source code in botree/s3.py 
28
29
30
31
32
33
34
35
36
37
38
39
def download(self, source: Path, target: Path, **kwargs):
    """
    Download a file from S3.

    Parameters
    ----------
    source : Path
        remote file path.
    target : Path
        local file path.
    """
    self.resource.download_file(str(source), str(target), **kwargs)
list_files(prefix: str = '', reverse: bool = False, *args: bool, **kwargs: bool) -> List[str]

List and sort (by date) all files in a given prefix.

Parameters:

  • prefix (str) –

    S3 prefix.

  • reverse (bool) –

    Reverse (descending) date sort, by default False

Returns:

  • List[str] –

    Paths to files.

Source code in botree/s3.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def list_files(
    self, prefix: str = "", reverse: bool = False, *args, **kwargs
) -> List[str]:
    """
    List and sort (by date) all files in a given prefix.

    Parameters
    ----------
    prefix : str
        S3 prefix.
    reverse : bool, optional
        Reverse (descending) date sort, by default False

    Returns
    -------
    List[str]
        Paths to files.
    """
    response = self.client.list_objects_v2(
        Bucket=self.name, Prefix=prefix, Delimiter="", *args, **kwargs
    )

    date_sorted = sorted(
        response["Contents"],  # type: ignore
        key=lambda x: x["LastModified"],  # type: ignore
        reverse=reverse,
    )

    keys = [key["Key"] for key in date_sorted]  # type: ignore

    return keys
list_folders(prefix: str = '', *args: str, **kwargs: str) -> List[str]

List all folders in a given prefix.

Parameters:

  • prefix (str) –

    S3 prefix.

Returns:

  • List[str] –

    Paths to folders.

Source code in botree/s3.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def list_folders(self, prefix: str = "", *args, **kwargs) -> List[str]:
    """
    List all folders in a given prefix.

    Parameters
    ----------
    prefix : str
        S3 prefix.

    Returns
    -------
    List[str]
        Paths to folders.
    """
    if not prefix.endswith("/"):
        prefix = prefix + "/"

    response = self.client.list_objects_v2(
        Bucket=self.name, Prefix=prefix, Delimiter="/", *args, **kwargs
    )

    keys = [key["Prefix"] for key in response["CommonPrefixes"]]  # type: ignore

    return keys
paginate_objects(prefix: str = '', page_size: int = 1000) -> Generator

Return a list all objects in a bucket with specified prefix.

Paginator is useful when you have 1000s of files in S3. S3 list_objects_v2 can list at max 1000 files in one go. :return: None

Source code in botree/s3.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def paginate_objects(self, prefix: str = "", page_size: int = 1000) -> Generator:
    """
    Return a list all objects in a bucket with specified prefix.

    Paginator is useful when you have 1000s of files in S3.
    S3 list_objects_v2 can list at max 1000 files in one go.
    :return: None
    """
    paginator = self.client.get_paginator("list_objects_v2")

    response = paginator.paginate(
        Bucket=self.name, Prefix=prefix, PaginationConfig={"PageSize": page_size}
    )

    for page in response:
        files = page.get("Contents")
        yield files
upload(source: Path, target: Path, **kwargs: Path)

Upload a file to S3.

Parameters:

  • source (Path) –

    local file path.

  • target (Path) –

    remote file path.

Source code in botree/s3.py 
41
42
43
44
45
46
47
48
49
50
51
52
def upload(self, source: Path, target: Path, **kwargs):
    """
    Upload a file to S3.

    Parameters
    ----------
    source : Path
        local file path.
    target : Path
        remote file path.
    """
    self.resource.upload_file(str(source), str(target), **kwargs)

botree.s3.S3(session: Session, **kwargs: Session)

AWS S3 operations.

S3 class init.

Source code in botree/s3.py 
180
181
182
183
def __init__(self, session: Session, **kwargs):
    """S3 class init."""
    self.session = session
    self.client = self.session.client(service_name="s3", **kwargs)

Functions

bucket(name: str, client_kwargs: dict = dict(), resource_kwargs: dict = dict()) -> Bucket

Get a bucket resource instance.

Source code in botree/s3.py 
194
195
196
197
198
199
200
201
def bucket(
    self,
    name: str,
    client_kwargs: dict = dict(),
    resource_kwargs: dict = dict(),
) -> Bucket:
    """Get a bucket resource instance."""
    return Bucket(self.session, name, **client_kwargs, **resource_kwargs)
create_bucket(name: str, *args: str, **kwargs: str)

Create a bucket.

Source code in botree/s3.py 
185
186
187
def create_bucket(self, name: str, *args, **kwargs):
    """Create a bucket."""
    self.client.create_bucket(Bucket=name, *args, **kwargs)
list_buckets() -> List[str]

Return a list of bucket names.

Source code in botree/s3.py 
189
190
191
192
def list_buckets(self) -> List[str]:
    """Return a list of bucket names."""
    response = self.client.list_buckets()
    return [bucket["Name"] for bucket in response["Buckets"]]  # type: ignore