Bases: FileManager
AWS S3 Bucket File manager.
This class provides an AWS S3 implementation of the FileManagerInterface,
the files handled by this class are stored in an AWS S3 bucket.
Set the env variable NUMEROUS_FILES_BACKEND to AWS_S3
to get an instance of this class when using the file_manager_factory method.
To save files on AWS S3 a bucket and a base prefix is needed to be specified
using the env variables NUMEROUS_FILES_BUCKET and NUMEROUS_FILES_BASE_PREFIX.
In order for the file_manager to access your files on aws,
you need to provide authentication to access AWS.
If you set the env variables NUMEROUS_FILES_AWS_ACCESS_KEY
and NUMEROUS_FILES_AWS_SECRET_KEY with your aws credentials
they will be used for authenticatication.
In case these env variables are not set the client will
authenticate with the AWS credentials from your environment.
If needed you can supply the names of the env variables you use to store
information of the bucket, base_prefix, aws_access and aws_secrets to
the file_manager_factory method as the following key word arguments:
bucket, base_prfix, aws_access_key_id, and aws_secret_access_key.
Source code in numerous/files/aws_s3.py
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
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
89
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
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139 | class FileManager(FileManagerInterface):
"""
AWS S3 Bucket File manager.
This class provides an AWS S3 implementation of the FileManagerInterface,
the files handled by this class are stored in an AWS S3 bucket.
Set the env variable `NUMEROUS_FILES_BACKEND` to `AWS_S3`
to get an instance of this class when using the file_manager_factory method.
To save files on AWS S3 a bucket and a base prefix is needed to be specified
using the env variables `NUMEROUS_FILES_BUCKET` and `NUMEROUS_FILES_BASE_PREFIX`.
In order for the file_manager to access your files on aws,
you need to provide authentication to access AWS.
If you set the env variables `NUMEROUS_FILES_AWS_ACCESS_KEY`
and `NUMEROUS_FILES_AWS_SECRET_KEY` with your aws credentials
they will be used for authenticatication.
In case these env variables are not set the client will
authenticate with the AWS credentials from your environment.
If needed you can supply the names of the env variables you use to store
information of the bucket, base_prefix, aws_access and aws_secrets to
the file_manager_factory method as the following key word arguments:
bucket, base_prfix, aws_access_key_id, and aws_secret_access_key.
"""
def __init__(self, bucket: str, base_prefix: str,
credentials:Dict[str,Any]|None=None) -> None:
self._bucket = bucket
self._base_prefix = base_prefix
if credentials is not None:
self._client = boto3.client("s3", **credentials)
else:
# Use the default session.
self._client = boto3.client("s3")
def put(self, src: StrOrPath, dst: str ) -> None:
"""
Upload a file to a path.
Args:
src: Source path.
dst: Destination path.
"""
self._client.upload_file(str(src), self._bucket, f"{self._base_prefix}/{dst}")
def remove(self, path: str) -> None:
"""
Remove a file at a path.
Args:
path: Path to file.
"""
self._client.delete_object(Bucket=self._bucket,
Key=f"{self._base_prefix}/{path}")
def list(self, path: str|None=None) -> List[str]:
"""
List files at a path.
Args:
path: Path to list files at.
"""
if path is None:
path = self._base_prefix
query_result = self._client.list_objects(
Bucket=self._bucket, Prefix=f"{self._base_prefix}/{path}")
if "Contents" not in query_result:
return []
list_results = [obj["Key"] for obj in query_result["Contents"]]
# Remove the base prefix from the results. But only first occurence.
return [result.replace(
f"{self._base_prefix}/", "", 1) for result in list_results]
def move(self, src: str, dst: str) -> None:
"""
Move a file from a source to a destination.
Args:
src: Source path.
dst: Destination path.
"""
self._client.copy_object(
Bucket=self._bucket,
CopySource=f"{self._bucket}/{self._base_prefix}/{src}",
Key=f"{self._base_prefix}/{dst}",
)
self._client.delete_object(Bucket=self._bucket,
Key=f"{self._base_prefix}/{src}")
def copy(self, src: str, dst: str) -> None:
"""
Copy a file from a source to a destination.
Args:
src: Source path.
dst: Destination path.
"""
self._client.copy_object(
Bucket=self._bucket,
CopySource=f"{self._bucket}/{self._base_prefix}/{src}",
Key=f"{self._base_prefix}/{dst}",
)
def get(self, src: str, dest: StrOrPath) -> None:
"""
Download a file from a source to a destination.
Args:
src: Source path.
dest: Destination path.
"""
self._client.download_file(
self._bucket,
f"{self._base_prefix}/{src}", str(dest),
)
|
copy(src, dst)
Copy a file from a source to a destination.
Source code in numerous/files/aws_s3.py
112
113
114
115
116
117
118
119
120
121
122
123
124
125 | def copy(self, src: str, dst: str) -> None:
"""
Copy a file from a source to a destination.
Args:
src: Source path.
dst: Destination path.
"""
self._client.copy_object(
Bucket=self._bucket,
CopySource=f"{self._bucket}/{self._base_prefix}/{src}",
Key=f"{self._base_prefix}/{dst}",
)
|
get(src, dest)
Download a file from a source to a destination.
| Parameters: |
-
src
(str)
–
-
dest
(StrOrPath)
–
|
Source code in numerous/files/aws_s3.py
127
128
129
130
131
132
133
134
135
136
137
138
139 | def get(self, src: str, dest: StrOrPath) -> None:
"""
Download a file from a source to a destination.
Args:
src: Source path.
dest: Destination path.
"""
self._client.download_file(
self._bucket,
f"{self._base_prefix}/{src}", str(dest),
)
|
list(path=None)
List files at a path.
| Parameters: |
-
path
(str | None, default:
None
)
–
|
Source code in numerous/files/aws_s3.py
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93 | def list(self, path: str|None=None) -> List[str]:
"""
List files at a path.
Args:
path: Path to list files at.
"""
if path is None:
path = self._base_prefix
query_result = self._client.list_objects(
Bucket=self._bucket, Prefix=f"{self._base_prefix}/{path}")
if "Contents" not in query_result:
return []
list_results = [obj["Key"] for obj in query_result["Contents"]]
# Remove the base prefix from the results. But only first occurence.
return [result.replace(
f"{self._base_prefix}/", "", 1) for result in list_results]
|
move(src, dst)
Move a file from a source to a destination.
Source code in numerous/files/aws_s3.py
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110 | def move(self, src: str, dst: str) -> None:
"""
Move a file from a source to a destination.
Args:
src: Source path.
dst: Destination path.
"""
self._client.copy_object(
Bucket=self._bucket,
CopySource=f"{self._bucket}/{self._base_prefix}/{src}",
Key=f"{self._base_prefix}/{dst}",
)
self._client.delete_object(Bucket=self._bucket,
Key=f"{self._base_prefix}/{src}")
|
put(src, dst)
Upload a file to a path.
| Parameters: |
-
src
(StrOrPath)
–
-
dst
(str)
–
|
Source code in numerous/files/aws_s3.py
52
53
54
55
56
57
58
59
60
61 | def put(self, src: StrOrPath, dst: str ) -> None:
"""
Upload a file to a path.
Args:
src: Source path.
dst: Destination path.
"""
self._client.upload_file(str(src), self._bucket, f"{self._base_prefix}/{dst}")
|
remove(path)
Remove a file at a path.
Source code in numerous/files/aws_s3.py
63
64
65
66
67
68
69
70
71
72 | def remove(self, path: str) -> None:
"""
Remove a file at a path.
Args:
path: Path to file.
"""
self._client.delete_object(Bucket=self._bucket,
Key=f"{self._base_prefix}/{path}")
|