Skip to content

Storage types

Bases: str

The file obect returned by the storage.

Source code in fastapi_storages/base.py 
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
class StorageFile(str):
    """
    The file obect returned by the storage.
    """

    def __new__(cls, name: str, storage: BaseStorage) -> "StorageFile":
        return str.__new__(cls, storage.get_path(name))

    def __init__(self, *, name: str, storage: BaseStorage):
        self._name = name
        self._storage = storage

    @property
    def name(self) -> str:
        """File name including extension."""

        return self._storage.get_name(self._name)

    @property
    def path(self) -> str:
        """Complete file path."""

        return self._storage.get_path(self._name)

    @property
    def size(self) -> int:
        """File size in bytes."""

        return self._storage.get_size(self._name)

    def open(self) -> BinaryIO:
        """
        Open a file handle of the file.
        """

        return self._storage.open(self._name)

    def write(self, file: BinaryIO) -> str:
        """
        Write input file which is opened in binary mode to destination.
        """

        if not self._storage.OVERWRITE_EXISTING_FILES:
            self._name = self._storage.generate_new_filename(self._name)

        return self._storage.write(file=file, name=self._name)

    def __str__(self) -> str:
        return self.path

name: str property

File name including extension.

path: str property

Complete file path.

size: int property

File size in bytes.

open()

Open a file handle of the file.

Source code in fastapi_storages/base.py 
58
59
60
61
62
63
def open(self) -> BinaryIO:
    """
    Open a file handle of the file.
    """

    return self._storage.open(self._name)

write(file)

Write input file which is opened in binary mode to destination.

Source code in fastapi_storages/base.py 
65
66
67
68
69
70
71
72
73
def write(self, file: BinaryIO) -> str:
    """
    Write input file which is opened in binary mode to destination.
    """

    if not self._storage.OVERWRITE_EXISTING_FILES:
        self._name = self._storage.generate_new_filename(self._name)

    return self._storage.write(file=file, name=self._name)

Bases: StorageFile

Inherits features of StorageFile and adds image specific properties.

Source code in fastapi_storages/base.py 
 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
class StorageImage(StorageFile):
    """
    Inherits features of `StorageFile` and adds image specific properties.
    """

    def __new__(
        cls, name: str, storage: BaseStorage, height: int, width: int
    ) -> "StorageImage":
        return str.__new__(cls, storage.get_path(name))

    def __init__(
        self, *, name: str, storage: BaseStorage, height: int, width: int
    ) -> None:
        super().__init__(name=name, storage=storage)
        self._width = width
        self._height = height

    @property
    def height(self) -> int:
        """
        Image height in pixels.
        """

        return self._height

    @property
    def width(self) -> int:
        """
        Image width in pixels.
        """

        return self._width

height: int property

Image height in pixels.

width: int property

Image width in pixels.

Bases: BaseStorage

File system storage which stores files in the local filesystem. You might want to use this with the FileType type.

Source code in fastapi_storages/filesystem.py 
 8
 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
class FileSystemStorage(BaseStorage):
    """
    File system storage which stores files in the local filesystem.
    You might want to use this with the `FileType` type.
    """

    default_chunk_size = 64 * 1024

    def __init__(self, path: str) -> None:
        self._path = Path(path)
        self._path.mkdir(parents=True, exist_ok=True)

    def get_name(self, name: str) -> str:
        """
        Get the normalized name of the file.
        """

        return secure_filename(Path(name).name)

    def get_path(self, name: str) -> str:
        """
        Get full path to the file.
        """

        return str(self._path / Path(name))

    def get_size(self, name: str) -> int:
        """
        Get file size in bytes.
        """

        return (self._path / name).stat().st_size

    def open(self, name: str) -> BinaryIO:
        """
        Open a file handle of the file object in binary mode.
        """

        path = self._path / Path(name)
        return open(path, "rb")

    def write(self, file: BinaryIO, name: str) -> str:
        """
        Write input file which is opened in binary mode to destination.
        """

        filename = secure_filename(name)
        path = self._path / Path(filename)

        file.seek(0, 0)
        with open(path, "wb") as output:
            while True:
                chunk = file.read(self.default_chunk_size)
                if not chunk:
                    break
                output.write(chunk)

        return str(path)

    def generate_new_filename(self, filename: str) -> str:
        counter = 0
        path = self._path / filename
        stem, extension = Path(filename).stem, Path(filename).suffix

        while path.exists():
            counter += 1
            path = self._path / f"{stem}_{counter}{extension}"

        return path.name

get_name(name)

Get the normalized name of the file.

Source code in fastapi_storages/filesystem.py 
20
21
22
23
24
25
def get_name(self, name: str) -> str:
    """
    Get the normalized name of the file.
    """

    return secure_filename(Path(name).name)

get_path(name)

Get full path to the file.

Source code in fastapi_storages/filesystem.py 
27
28
29
30
31
32
def get_path(self, name: str) -> str:
    """
    Get full path to the file.
    """

    return str(self._path / Path(name))

get_size(name)

Get file size in bytes.

Source code in fastapi_storages/filesystem.py 
34
35
36
37
38
39
def get_size(self, name: str) -> int:
    """
    Get file size in bytes.
    """

    return (self._path / name).stat().st_size

open(name)

Open a file handle of the file object in binary mode.

Source code in fastapi_storages/filesystem.py 
41
42
43
44
45
46
47
def open(self, name: str) -> BinaryIO:
    """
    Open a file handle of the file object in binary mode.
    """

    path = self._path / Path(name)
    return open(path, "rb")

write(file, name)

Write input file which is opened in binary mode to destination.

Source code in fastapi_storages/filesystem.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def write(self, file: BinaryIO, name: str) -> str:
    """
    Write input file which is opened in binary mode to destination.
    """

    filename = secure_filename(name)
    path = self._path / Path(filename)

    file.seek(0, 0)
    with open(path, "wb") as output:
        while True:
            chunk = file.read(self.default_chunk_size)
            if not chunk:
                break
            output.write(chunk)

    return str(path)

Bases: BaseStorage

Amazon S3 or any S3 compatible storage backend. You might want to use this with the FileType type. Requires boto3 to be installed.

Source code in fastapi_storages/s3.py 
 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
140
141
142
143
144
class S3Storage(BaseStorage):
    """
    Amazon S3 or any S3 compatible storage backend.
    You might want to use this with the `FileType` type.
    Requires `boto3` to be installed.
    """

    default_content_type = "application/octet-stream"

    AWS_ACCESS_KEY_ID = os.environ.get("AWS_ACCESS_KEY_ID", "")
    """AWS access key ID. Either set here or as an environment variable."""

    AWS_SECRET_ACCESS_KEY = os.environ.get("AWS_SECRET_ACCESS_KEY", "")
    """AWS secret access key. Either set here or as an environment variable."""

    AWS_S3_BUCKET_NAME = ""
    """AWS S3 bucket name to use."""

    AWS_S3_ENDPOINT_URL = ""
    """AWS S3 endpoint URL."""

    AWS_S3_USE_SSL = True
    """Indicate if SSL should be used."""

    AWS_DEFAULT_ACL = ""
    """Optional ACL set on the object like `public-read`.
    By default file will be private."""

    AWS_QUERYSTRING_AUTH = False
    """Indicate if query parameter authentication should be used in URLs."""

    AWS_S3_CUSTOM_DOMAIN = ""
    """Custom domain to use for serving object URLs."""

    def __init__(self) -> None:
        assert boto3 is not None, "'boto3' is not installed"
        assert not self.AWS_S3_ENDPOINT_URL.startswith(
            "http"
        ), "URL should not contain protocol"

        self._http_scheme = "https" if self.AWS_S3_USE_SSL else "http"
        self._url = f"{self._http_scheme}://{self.AWS_S3_ENDPOINT_URL}"
        self._s3 = boto3.resource(
            "s3",
            endpoint_url=self._url,
            use_ssl=self.AWS_S3_USE_SSL,
            aws_access_key_id=self.AWS_ACCESS_KEY_ID,
            aws_secret_access_key=self.AWS_SECRET_ACCESS_KEY,
        )
        self._bucket = self._s3.Bucket(name=self.AWS_S3_BUCKET_NAME)

    def get_name(self, name: str) -> str:
        """
        Get the normalized name of the file.
        """

        filename = secure_filename(Path(name).name)
        return str(Path(name).with_name(filename))

    def get_path(self, name: str) -> str:
        """
        Get full URL to the file.
        """

        key = self.get_name(name)

        if self.AWS_S3_CUSTOM_DOMAIN:
            return "{}://{}/{}".format(
                self._http_scheme,
                self.AWS_S3_CUSTOM_DOMAIN,
                key,
            )

        if self.AWS_QUERYSTRING_AUTH:
            params = {"Bucket": self._bucket.name, "Key": key}
            return self._s3.meta.client.generate_presigned_url(
                "get_object", Params=params
            )

        return "{}://{}/{}/{}".format(
            self._http_scheme,
            self.AWS_S3_ENDPOINT_URL,
            self.AWS_S3_BUCKET_NAME,
            key,
        )

    def get_size(self, name: str) -> int:
        """
        Get file size in bytes.
        """

        key = self.get_name(name)
        return self._bucket.Object(key).content_length

    def write(self, file: BinaryIO, name: str) -> str:
        """
        Write input file which is opened in binary mode to destination.
        """

        file.seek(0, 0)
        key = self.get_name(name)
        content_type, _ = mimetypes.guess_type(key)
        params = {
            "ACL": self.AWS_DEFAULT_ACL,
            "ContentType": content_type or self.default_content_type,
        }
        self._bucket.upload_fileobj(file, key, ExtraArgs=params)
        return key

    def generate_new_filename(self, filename: str) -> str:
        key = self.get_name(filename)
        stem = Path(filename).stem
        suffix = Path(filename).suffix
        counter = 0

        while self._check_object_exists(key):
            counter += 1
            filename = f"{stem}_{counter}{suffix}"
            key = self.get_name(filename)

        return filename

    def _check_object_exists(self, key: str) -> bool:
        try:
            self._bucket.Object(key).load()
        except boto3.exceptions.botocore.exceptions.ClientError as e:
            if e.response["Error"]["Code"] == "404":
                return False

        return True

AWS_ACCESS_KEY_ID = os.environ.get('AWS_ACCESS_KEY_ID', '') class-attribute instance-attribute

AWS access key ID. Either set here or as an environment variable.

AWS_DEFAULT_ACL = '' class-attribute instance-attribute

Optional ACL set on the object like public-read. By default file will be private.

AWS_QUERYSTRING_AUTH = False class-attribute instance-attribute

Indicate if query parameter authentication should be used in URLs.

AWS_S3_BUCKET_NAME = '' class-attribute instance-attribute

AWS S3 bucket name to use.

AWS_S3_CUSTOM_DOMAIN = '' class-attribute instance-attribute

Custom domain to use for serving object URLs.

AWS_S3_ENDPOINT_URL = '' class-attribute instance-attribute

AWS S3 endpoint URL.

AWS_S3_USE_SSL = True class-attribute instance-attribute

Indicate if SSL should be used.

AWS_SECRET_ACCESS_KEY = os.environ.get('AWS_SECRET_ACCESS_KEY', '') class-attribute instance-attribute

AWS secret access key. Either set here or as an environment variable.

get_name(name)

Get the normalized name of the file.

Source code in fastapi_storages/s3.py 
66
67
68
69
70
71
72
def get_name(self, name: str) -> str:
    """
    Get the normalized name of the file.
    """

    filename = secure_filename(Path(name).name)
    return str(Path(name).with_name(filename))

get_path(name)

Get full URL to the file.

Source code in fastapi_storages/s3.py 
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
def get_path(self, name: str) -> str:
    """
    Get full URL to the file.
    """

    key = self.get_name(name)

    if self.AWS_S3_CUSTOM_DOMAIN:
        return "{}://{}/{}".format(
            self._http_scheme,
            self.AWS_S3_CUSTOM_DOMAIN,
            key,
        )

    if self.AWS_QUERYSTRING_AUTH:
        params = {"Bucket": self._bucket.name, "Key": key}
        return self._s3.meta.client.generate_presigned_url(
            "get_object", Params=params
        )

    return "{}://{}/{}/{}".format(
        self._http_scheme,
        self.AWS_S3_ENDPOINT_URL,
        self.AWS_S3_BUCKET_NAME,
        key,
    )

get_size(name)

Get file size in bytes.

Source code in fastapi_storages/s3.py 
101
102
103
104
105
106
107
def get_size(self, name: str) -> int:
    """
    Get file size in bytes.
    """

    key = self.get_name(name)
    return self._bucket.Object(key).content_length

write(file, name)

Write input file which is opened in binary mode to destination.

Source code in fastapi_storages/s3.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def write(self, file: BinaryIO, name: str) -> str:
    """
    Write input file which is opened in binary mode to destination.
    """

    file.seek(0, 0)
    key = self.get_name(name)
    content_type, _ = mimetypes.guess_type(key)
    params = {
        "ACL": self.AWS_DEFAULT_ACL,
        "ContentType": content_type or self.default_content_type,
    }
    self._bucket.upload_fileobj(file, key, ExtraArgs=params)
    return key