AWS File Manager

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.

Parameters:
  • src (str) –

    Source path.

  • dst (str) –

    Destination path.

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) –

    Source path.

  • dest (StrOrPath) –

    Destination path.

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 ) –

    Path to list files at.

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.

Parameters:
  • src (str) –

    Source path.

  • dst (str) –

    Destination path.

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) –

    Source path.

  • dst (str) –

    Destination path.

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.

Parameters:
  • path (str) –

    Path to file.

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}")