Skip to content

Cloudflare Images

Cloudflare Images is a Paid Service

Pricing:

  • $5 per month per 100k images stored
  • $1 per month per 100k images delivered

API v1

Bases: CF

Need to setup a Cloudflare Images account to use. See Cloudflare Images docs. With required variables secured:

Field in .env Cloudflare API Credential Where credential found
CF_ACCT_ID Account ID https://dash.cloudflare.com/<acct_id>/images/images
CF_IMG_HASH Account Hash https://dash.cloudflare.com/<acct_id>/images/images
CF_IMG_TOKEN API Secret Generate / save via https://dash.cloudflare.com/<acct_id>/profile/api-tokens

Add secrets to .env file and use as follows:

Examples:

Example Usage
 1
 2
 3
 4
 5
 6
 7
 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
>>> cf = CloudflareImagesAPIv1() # will error out since missing key values
Traceback (most recent call last):
pydantic_core._pydantic_core.ValidationError: 3 validation errors for CloudflareImagesAPIv1
CF_ACCT_ID
  Field required [type=missing, input_value={}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.3/v/missing
CF_IMG_HASH
  Field required [type=missing, input_value={}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.3/v/missing
CF_IMG_TOKEN
  Field required [type=missing, input_value={}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.3/v/missing
>>> import os
>>> os.environ['CF_ACCT_ID'] = "ABC"
>>> cf = CloudflareImagesAPIv1() # will error out since still missing other values
Traceback (most recent call last):
pydantic_core._pydantic_core.ValidationError: 2 validation errors for CloudflareImagesAPIv1
CF_IMG_HASH
  Field required [type=missing, input_value={'CF_ACCT_ID': 'ABC'}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.3/v/missing
CF_IMG_TOKEN
  Field required [type=missing, input_value={'CF_ACCT_ID': 'ABC'}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.3/v/missing
>>> # we'll add all the values needed
>>> os.environ['CF_IMG_HASH'], os.environ['CF_IMG_TOKEN'] = "DEF", "XYZ"
>>> cf = CloudflareImagesAPIv1() # no longer errors out
>>> CF.set_bearer_auth(cf.api_token)
{'Authorization': 'Bearer XYZ'}
>>> cf.base_api
'https://api.cloudflare.com/client/v4/accounts/ABC/images/v1'
>>> cf.base_delivery
'https://imagedelivery.net/DEF'
>>> cf.url('hi-bob', 'w=400,sharpen=3')
'https://imagedelivery.net/DEF/hi-bob/w=400,sharpen=3'
>>> from pathlib import Path
>>> p = Path().cwd() / "img" / "screenshot.png"
>>> p.exists() # Sample image found in `/img/screenshot.png`
True
>>> import io
>>> img = io.BytesIO(p.read_bytes())
>>> type(img)
<class '_io.BytesIO'>
>>> # Can now use img in `cf.post('sample_id', img)`
Source code in cloudflare_images/api.py
Python
 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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
class CloudflareImagesAPIv1(CF):
    """
    Need to setup a Cloudflare Images account to use. See Cloudflare Images [docs](https://developers.cloudflare.com/images/cloudflare-images/).
    With required variables secured:

    Field in .env | Cloudflare API Credential | Where credential found
    :--|:--:|:--
    `CF_ACCT_ID` | Account ID |  `https://dash.cloudflare.com/<acct_id>/images/images`
    `CF_IMG_HASH` | Account Hash | `https://dash.cloudflare.com/<acct_id>/images/images`
    `CF_IMG_TOKEN` | API Secret | Generate / save via `https://dash.cloudflare.com/<acct_id>/profile/api-tokens`

    Add secrets to .env file and use as follows:

    Examples:
        ```py title="Example Usage" linenums="1"
        >>> cf = CloudflareImagesAPIv1() # will error out since missing key values
        Traceback (most recent call last):
        pydantic_core._pydantic_core.ValidationError: 3 validation errors for CloudflareImagesAPIv1
        CF_ACCT_ID
          Field required [type=missing, input_value={}, input_type=dict]
            For further information visit https://errors.pydantic.dev/2.3/v/missing
        CF_IMG_HASH
          Field required [type=missing, input_value={}, input_type=dict]
            For further information visit https://errors.pydantic.dev/2.3/v/missing
        CF_IMG_TOKEN
          Field required [type=missing, input_value={}, input_type=dict]
            For further information visit https://errors.pydantic.dev/2.3/v/missing
        >>> import os
        >>> os.environ['CF_ACCT_ID'] = "ABC"
        >>> cf = CloudflareImagesAPIv1() # will error out since still missing other values
        Traceback (most recent call last):
        pydantic_core._pydantic_core.ValidationError: 2 validation errors for CloudflareImagesAPIv1
        CF_IMG_HASH
          Field required [type=missing, input_value={'CF_ACCT_ID': 'ABC'}, input_type=dict]
            For further information visit https://errors.pydantic.dev/2.3/v/missing
        CF_IMG_TOKEN
          Field required [type=missing, input_value={'CF_ACCT_ID': 'ABC'}, input_type=dict]
            For further information visit https://errors.pydantic.dev/2.3/v/missing
        >>> # we'll add all the values needed
        >>> os.environ['CF_IMG_HASH'], os.environ['CF_IMG_TOKEN'] = "DEF", "XYZ"
        >>> cf = CloudflareImagesAPIv1() # no longer errors out
        >>> CF.set_bearer_auth(cf.api_token)
        {'Authorization': 'Bearer XYZ'}
        >>> cf.base_api
        'https://api.cloudflare.com/client/v4/accounts/ABC/images/v1'
        >>> cf.base_delivery
        'https://imagedelivery.net/DEF'
        >>> cf.url('hi-bob', 'w=400,sharpen=3')
        'https://imagedelivery.net/DEF/hi-bob/w=400,sharpen=3'
        >>> from pathlib import Path
        >>> p = Path().cwd() / "img" / "screenshot.png"
        >>> p.exists() # Sample image found in `/img/screenshot.png`
        True
        >>> import io
        >>> img = io.BytesIO(p.read_bytes())
        >>> type(img)
        <class '_io.BytesIO'>
        >>> # Can now use img in `cf.post('sample_id', img)`
        ```
    """  # noqa: E501

    account_id: str = Field(
        default=...,
        repr=False,
        title="Cloudflare Account ID",
        description="Overrides the base setting by making this mandatory.",
        validation_alias="CF_ACCT_ID",
    )
    cf_img_hash: str = Field(
        default=...,
        repr=False,
        title="Cloudflare Image Hash",
        description="Assigned when you create a Cloudflare Images account",
        validation_alias="CF_IMG_HASH",
    )
    api_token: str = Field(
        default=...,
        repr=False,
        title="Cloudflare Image API Token",
        description="Secure token to perform API operations",
        validation_alias="CF_IMG_TOKEN",
    )
    client_api_ver: str = Field(
        default="v4",
        title="Cloudflare Client API Version",
        description="Used in the middle of the URL in API requests.",
        validation_alias="CLOUDFLARE_CLIENT_API_VERSION",
    )
    images_api_ver: str = Field(
        default="v1",
        title="Cloudflare Images API Version",
        description="Used at the end of URL in API requests.",
        validation_alias="CLOUDFLARE_IMAGES_API_VERSION",
    )
    timeout: int = Field(
        default=60,
        validation_alias="CF_IMG_TOKEN_TIMEOUT",
    )

    @property
    def client(self):
        return httpx.Client(timeout=self.timeout)

    @property
    def base_api(self) -> str:
        """Construct endpoint. See [formula](https://developers.cloudflare.com/images/cloudflare-images/api-request/).

        Examples:
            >>> import os
            >>> os.environ['CF_ACCT_ID'] = "ABC"
            >>> os.environ['CF_IMG_HASH'], os.environ['CF_IMG_TOKEN'] = "DEF", "XYZ"
            >>> cf = CloudflareImagesAPIv1()
            >>> cf.base_api
            'https://api.cloudflare.com/client/v4/accounts/ABC/images/v1'

        Returns:
            str: URL endpoint to make requests with the Cloudflare-supplied credentials.
        """
        return self.add_account_endpoint(
            f"/{self.account_id}/images/{self.images_api_ver}"
        )

    @property
    def v2(self) -> str:
        """See updated [list API endpoint](https://developers.cloudflare.com/api/operations/cloudflare-images-list-images-v2).

        Examples:
            >>> import os
            >>> os.environ['CF_ACCT_ID'] = "ABC"
            >>> os.environ['CF_IMG_HASH'], os.environ['CF_IMG_TOKEN'] = "DEF", "XYZ"
            >>> cf = CloudflareImagesAPIv1()
            >>> cf.v2_api
            'https://api.cloudflare.com/client/v4/accounts/ABC/images/v2'
        """
        return self.add_account_endpoint(f"/{self.account_id}/images/v2")

    @property
    def base_delivery(self):
        """Images are served with the following format: `https://imagedelivery.net/<ACCOUNT_HASH>/<IMAGE_ID>/<VARIANT_NAME>`

        This property constructs the first part: `https://imagedelivery.net/<ACCOUNT_HASH>`

        See Cloudflare [docs](https://developers.cloudflare.com/images/cloudflare-images/serve-images/).

        Examples:
        >>> import os
            >>> os.environ['CF_ACCT_ID'] = "ABC"
            >>> os.environ['CF_IMG_HASH'], os.environ['CF_IMG_TOKEN'] = "DEF", "XYZ"
            >>> cf = CloudflareImagesAPIv1()
            >>> cf.base_delivery
            'https://imagedelivery.net/DEF'
        """  # noqa: E501
        return "/".join([CF_DELIVER, self.cf_img_hash])

    def url(self, img_id: str, variant: str = "public") -> str:
        """Generates url based on the Cloudflare hash of the account. The `variant` is based on
        how these are customized on Cloudflare Images. See also flexible variant [docs](https://developers.cloudflare.com/images/cloudflare-images/transform/flexible-variants/)

        Examples:
            >>> import os
            >>> os.environ['CF_ACCT_ID'] = "ABC"
            >>> os.environ['CF_IMG_HASH'], os.environ['CF_IMG_TOKEN'] = "DEF", "XYZ"
            >>> cf = CloudflareImagesAPIv1()
            >>> cf.url('sample-img', 'avatar')
            'https://imagedelivery.net/DEF/sample-img/avatar'

        Args:
            img_id (str): The uploaded ID
            variant (str, optional): The variant created in the Cloudflare Images dashboard. Defaults to "public".

        Returns:
            str: URL to display the request `img_id` with `variant`.
        """  # noqa: E501
        return "/".join([self.base_delivery, img_id, variant])

    def get(self, img_id: str, *args, **kwargs) -> httpx.Response:
        """Issue httpx GET request to the image found in storage. Assuming request like
        `CFImage().get('target-img-id')`, returns a response with metadata:

        Examples:
            ```py title="Response object from Cloudflare Images"
            >>> # CFImage().get('target-img-id') commented out since hypothetical
            b'{
                "result": {
                    "id": "target-img-id",
                    "filename": "target-img-id",
                    "uploaded": "2023-02-20T09:09:41.755Z",
                    "requireSignedURLs": false,
                    "variants": [
                        "https://imagedelivery.net/<hash>/<target-img-id>/public",
                        "https://imagedelivery.net/<hash>/<target-img-id>/cover",
                        "https://imagedelivery.net/<hash>/<target-img-id>/avatar",
                        "https://imagedelivery.net/<hash>/<target-img-id>/uniform"
                    ]
                },
                "success": true,
                "errors": [],
                "messages": []
            }'
            ```
        """
        return self.client.get(
            url=f"{self.base_api}/{img_id}",
            headers=CF.set_bearer_auth(self.api_token),
            *args,
            **kwargs,
        )

    def delete(self, img_id: str, *args, **kwargs) -> httpx.Response:
        """Issue httpx [DELETE](https://developers.cloudflare.com/images/cloudflare-images/transform/delete-images/) request to the image."""  # noqa: E501
        return self.client.delete(
            url=f"{self.base_api}/{img_id}",
            headers=CF.set_bearer_auth(self.api_token),
            *args,
            **kwargs,
        )

    def post(self, img_id: str, img: bytes, *args, **kwargs) -> httpx.Response:
        """Issue httpx [POST](https://developers.cloudflare.com/images/cloudflare-images/upload-images/upload-via-url/) request to upload image."""  # noqa: E501
        return self.client.post(
            url=self.base_api,
            headers=CF.set_bearer_auth(self.api_token),
            data={"id": img_id},
            files={"file": (img_id, img)},
            *args,
            **kwargs,
        )

    def upsert(self, img_id: str, img: bytes) -> httpx.Response:
        """Ensures a unique id name by first deleting the `img_id` from storage and then
        uploading the `img`."""
        self.delete(img_id)
        return self.post(img_id, img)

    def list_images(
        self,
        per_page: int = 1000,
        sort_order: str = "desc",
        continuation_token: str | None = None,
    ) -> httpx.Response:
        """See [list images API](https://developers.cloudflare.com/api/operations/cloudflare-images-list-images-v2).

        Args:
            per_page (int, optional): Number of items per page (10 to 10,000). Defaults to 1000.
            sort_order (str, optional): Sorting order by upload time (asc | desc). Defaults to "desc".
            continuation_token (str | None, optional): Continuation token for a next page. List images V2 returns continuation_token. Defaults to None.

        Returns:
            httpx.Response: Contains top-level fields for `success`, `errors`, `messages` and the `result`.
        """  # noqa: E501

        if per_page < 10 or per_page > 10000:
            raise Exception(f"Improper {per_page=}")
        if sort_order not in ["asc", "desc"]:
            raise Exception(f"Improper {sort_order=}")
        params = {"per_page": per_page, "sort_order": sort_order}

        if continuation_token:
            params["continuation_token"] = continuation_token
        qs = urlencode(params)

        return self.client.get(
            url=f"{self.v2}?{qs}", headers=CF.set_bearer_auth(self.api_token)
        )

Attributes

base_api: str property

Construct endpoint. See formula.

Examples:

Python Console Session
>>> import os
>>> os.environ['CF_ACCT_ID'] = "ABC"
>>> os.environ['CF_IMG_HASH'], os.environ['CF_IMG_TOKEN'] = "DEF", "XYZ"
>>> cf = CloudflareImagesAPIv1()
>>> cf.base_api
'https://api.cloudflare.com/client/v4/accounts/ABC/images/v1'

Returns:

Name Type Description
str str

URL endpoint to make requests with the Cloudflare-supplied credentials.

base_delivery property

Images are served with the following format: https://imagedelivery.net/<ACCOUNT_HASH>/<IMAGE_ID>/<VARIANT_NAME>

This property constructs the first part: https://imagedelivery.net/<ACCOUNT_HASH>

See Cloudflare docs.

Examples:

import os >>> os.environ['CF_ACCT_ID'] = "ABC" >>> os.environ['CF_IMG_HASH'], os.environ['CF_IMG_TOKEN'] = "DEF", "XYZ" >>> cf = CloudflareImagesAPIv1() >>> cf.base_delivery 'https://imagedelivery.net/DEF'

v2: str property

See updated list API endpoint.

Examples:

Python Console Session
>>> import os
>>> os.environ['CF_ACCT_ID'] = "ABC"
>>> os.environ['CF_IMG_HASH'], os.environ['CF_IMG_TOKEN'] = "DEF", "XYZ"
>>> cf = CloudflareImagesAPIv1()
>>> cf.v2_api
'https://api.cloudflare.com/client/v4/accounts/ABC/images/v2'

Functions

delete(img_id, *args, **kwargs)

Issue httpx DELETE request to the image.

Source code in cloudflare_images/api.py
Python
219
220
221
222
223
224
225
226
def delete(self, img_id: str, *args, **kwargs) -> httpx.Response:
    """Issue httpx [DELETE](https://developers.cloudflare.com/images/cloudflare-images/transform/delete-images/) request to the image."""  # noqa: E501
    return self.client.delete(
        url=f"{self.base_api}/{img_id}",
        headers=CF.set_bearer_auth(self.api_token),
        *args,
        **kwargs,
    )

get(img_id, *args, **kwargs)

Issue httpx GET request to the image found in storage. Assuming request like CFImage().get('target-img-id'), returns a response with metadata:

Examples:

Response object from Cloudflare Images
>>> # CFImage().get('target-img-id') commented out since hypothetical
b'{
    "result": {
        "id": "target-img-id",
        "filename": "target-img-id",
        "uploaded": "2023-02-20T09:09:41.755Z",
        "requireSignedURLs": false,
        "variants": [
            "https://imagedelivery.net/<hash>/<target-img-id>/public",
            "https://imagedelivery.net/<hash>/<target-img-id>/cover",
            "https://imagedelivery.net/<hash>/<target-img-id>/avatar",
            "https://imagedelivery.net/<hash>/<target-img-id>/uniform"
        ]
    },
    "success": true,
    "errors": [],
    "messages": []
}'
Source code in cloudflare_images/api.py
Python
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def get(self, img_id: str, *args, **kwargs) -> httpx.Response:
    """Issue httpx GET request to the image found in storage. Assuming request like
    `CFImage().get('target-img-id')`, returns a response with metadata:

    Examples:
        ```py title="Response object from Cloudflare Images"
        >>> # CFImage().get('target-img-id') commented out since hypothetical
        b'{
            "result": {
                "id": "target-img-id",
                "filename": "target-img-id",
                "uploaded": "2023-02-20T09:09:41.755Z",
                "requireSignedURLs": false,
                "variants": [
                    "https://imagedelivery.net/<hash>/<target-img-id>/public",
                    "https://imagedelivery.net/<hash>/<target-img-id>/cover",
                    "https://imagedelivery.net/<hash>/<target-img-id>/avatar",
                    "https://imagedelivery.net/<hash>/<target-img-id>/uniform"
                ]
            },
            "success": true,
            "errors": [],
            "messages": []
        }'
        ```
    """
    return self.client.get(
        url=f"{self.base_api}/{img_id}",
        headers=CF.set_bearer_auth(self.api_token),
        *args,
        **kwargs,
    )

list_images(per_page=1000, sort_order='desc', continuation_token=None)

See list images API.

Parameters:

Name Type Description Default
per_page int

Number of items per page (10 to 10,000). Defaults to 1000.

 1000
sort_order str

Sorting order by upload time (asc | desc). Defaults to "desc".

 'desc'
continuation_token str | None

Continuation token for a next page. List images V2 returns continuation_token. Defaults to None.

 None

Returns:

Type Description
Response

httpx.Response: Contains top-level fields for success, errors, messages and the result.
Source code in cloudflare_images/api.py
Python
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
def list_images(
    self,
    per_page: int = 1000,
    sort_order: str = "desc",
    continuation_token: str | None = None,
) -> httpx.Response:
    """See [list images API](https://developers.cloudflare.com/api/operations/cloudflare-images-list-images-v2).

    Args:
        per_page (int, optional): Number of items per page (10 to 10,000). Defaults to 1000.
        sort_order (str, optional): Sorting order by upload time (asc | desc). Defaults to "desc".
        continuation_token (str | None, optional): Continuation token for a next page. List images V2 returns continuation_token. Defaults to None.

    Returns:
        httpx.Response: Contains top-level fields for `success`, `errors`, `messages` and the `result`.
    """  # noqa: E501

    if per_page < 10 or per_page > 10000:
        raise Exception(f"Improper {per_page=}")
    if sort_order not in ["asc", "desc"]:
        raise Exception(f"Improper {sort_order=}")
    params = {"per_page": per_page, "sort_order": sort_order}

    if continuation_token:
        params["continuation_token"] = continuation_token
    qs = urlencode(params)

    return self.client.get(
        url=f"{self.v2}?{qs}", headers=CF.set_bearer_auth(self.api_token)
    )

post(img_id, img, *args, **kwargs)

Issue httpx POST request to upload image.

Source code in cloudflare_images/api.py
Python
228
229
230
231
232
233
234
235
236
237
def post(self, img_id: str, img: bytes, *args, **kwargs) -> httpx.Response:
    """Issue httpx [POST](https://developers.cloudflare.com/images/cloudflare-images/upload-images/upload-via-url/) request to upload image."""  # noqa: E501
    return self.client.post(
        url=self.base_api,
        headers=CF.set_bearer_auth(self.api_token),
        data={"id": img_id},
        files={"file": (img_id, img)},
        *args,
        **kwargs,
    )

upsert(img_id, img)

Ensures a unique id name by first deleting the img_id from storage and then uploading the img.

Source code in cloudflare_images/api.py
Python
239
240
241
242
243
def upsert(self, img_id: str, img: bytes) -> httpx.Response:
    """Ensures a unique id name by first deleting the `img_id` from storage and then
    uploading the `img`."""
    self.delete(img_id)
    return self.post(img_id, img)

url(img_id, variant='public')

Generates url based on the Cloudflare hash of the account. The variant is based on how these are customized on Cloudflare Images. See also flexible variant docs

Examples:

Python Console Session
>>> import os
>>> os.environ['CF_ACCT_ID'] = "ABC"
>>> os.environ['CF_IMG_HASH'], os.environ['CF_IMG_TOKEN'] = "DEF", "XYZ"
>>> cf = CloudflareImagesAPIv1()
>>> cf.url('sample-img', 'avatar')
'https://imagedelivery.net/DEF/sample-img/avatar'

Parameters:

Name Type Description Default
img_id str

The uploaded ID

 required
variant str

The variant created in the Cloudflare Images dashboard. Defaults to "public".

 'public'

Returns:

Name Type Description
str str

URL to display the request img_id with variant.
Source code in cloudflare_images/api.py
Python
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def url(self, img_id: str, variant: str = "public") -> str:
    """Generates url based on the Cloudflare hash of the account. The `variant` is based on
    how these are customized on Cloudflare Images. See also flexible variant [docs](https://developers.cloudflare.com/images/cloudflare-images/transform/flexible-variants/)

    Examples:
        >>> import os
        >>> os.environ['CF_ACCT_ID'] = "ABC"
        >>> os.environ['CF_IMG_HASH'], os.environ['CF_IMG_TOKEN'] = "DEF", "XYZ"
        >>> cf = CloudflareImagesAPIv1()
        >>> cf.url('sample-img', 'avatar')
        'https://imagedelivery.net/DEF/sample-img/avatar'

    Args:
        img_id (str): The uploaded ID
        variant (str, optional): The variant created in the Cloudflare Images dashboard. Defaults to "public".

    Returns:
        str: URL to display the request `img_id` with `variant`.
    """  # noqa: E501
    return "/".join([self.base_delivery, img_id, variant])

Django

Bases: Storage

Custom Storage Class based on Django docs instructions

Starting with Django 4.2, add to STORAGES setting:

Django settings.py
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
...
STORAGES = {  # django 4.2 and above
    "default": {  # default
        "BACKEND": "django.core.files.storage.FileSystemStorage",
    },
    "staticfiles": {  # default
        "BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage",
    },
    "cloudflare_images": {  # add
        "BACKEND": "cloudflare_images.django.ImageStorageCloudflare",
    },
}

Can then define a callable likeso:

For use in ImageField
from django.core.files.storage import storages


def select_storage(is_remote_env: bool):
    return storages["cloudflare_images"] if is_remote_env else storages["default"]


class MyModel(models.Model):
    my_img = models.ImageField(storage=select_storage)

Can also refer to it via:

Invocation
from django.core.files.storage import storages
cf = storages["cloudflare_images"]

# assume previous upload done
id = <image-id-uploaded>

# get image url, defaults to 'public' variant
cf.url(id)

# specified 'avatar' variant, assuming it was created in the Cloudflare Images dashboard / API
cf.url_variant(id, 'avatar')
Source code in cloudflare_images/dj.py
Python
 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
@deconstructible
class LimitedStorageCloudflareImages(Storage):
    """Custom Storage Class based on Django docs [instructions](https://docs.djangoproject.com/en/dev/howto/custom-file-storage/#django.core.files.storage._open)

    Starting with Django 4.2, add to `STORAGES` setting:

    ```python title="Django settings.py" linenums="1" hl_lines="9 10"
    ...
    STORAGES = {  # django 4.2 and above
        "default": {  # default
            "BACKEND": "django.core.files.storage.FileSystemStorage",
        },
        "staticfiles": {  # default
            "BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage",
        },
        "cloudflare_images": {  # add
            "BACKEND": "cloudflare_images.django.ImageStorageCloudflare",
        },
    }
    ```

    Can then define a [callable](https://docs.djangoproject.com/en/dev/topics/files/#using-a-callable) likeso:

    ```python title="For use in ImageField"
    from django.core.files.storage import storages


    def select_storage(is_remote_env: bool):
        return storages["cloudflare_images"] if is_remote_env else storages["default"]


    class MyModel(models.Model):
        my_img = models.ImageField(storage=select_storage)
    ```

    Can also refer to it via:

    ```python title="Invocation"
    from django.core.files.storage import storages
    cf = storages["cloudflare_images"]

    # assume previous upload done
    id = <image-id-uploaded>

    # get image url, defaults to 'public' variant
    cf.url(id)

    # specified 'avatar' variant, assuming it was created in the Cloudflare Images dashboard / API
    cf.url_variant(id, 'avatar')
    ```


    """  # noqa: E501

    def __init__(self):
        super().__init__()
        self.api = CloudflareImagesAPIv1()

    def __repr__(self):
        return "<LimitedToImagesStorageClassCloudflare>"

    def _open(self, name: str, mode="rb") -> File:
        return File(self.api.get(img_id=name), name=name)

    def _save(self, name: str, content: bytes) -> str:
        res = self.api.upsert(name, content)
        return self.api.url(img_id=res.json()["result"]["id"])

    def get_valid_name(self, name):
        return name

    def get_available_name(self, name, max_length=None):
        return self.generate_filename(name)

    def generate_filename(self, filename):
        return filename

    def delete(self, name) -> httpx.Response:
        return self.api.delete(name)

    def exists(self, name: str) -> bool:
        res = self.api.get(name)
        if res.status_code == HTTPStatus.NOT_FOUND:
            return False
        elif res.status_code == HTTPStatus.OK:
            return True
        raise Exception("Image name found but http status code is not OK.")

    def listdir(self, path):
        raise NotImplementedError(
            "subclasses of Storage must provide a listdir() method"
        )

    def size(self, name: str):
        return len(self.api.get(name).content)

    def url(self, name: str):
        return self.api.url(name)

    def url_variant(self, name: str, variant: str):
        return self.api.url(name, variant)

    def get_accessed_time(self, name):
        raise NotImplementedError(
            "subclasses of Storage must provide a get_accessed_time() method"
        )

    def get_created_time(self, name):
        raise NotImplementedError(
            "subclasses of Storage must provide a get_created_time() method"
        )

    def get_modified_time(self, name):
        raise NotImplementedError(
            "subclasses of Storage must provide a get_modified_time() method"
        )