Instantly share code, notes, and snippets.

What would you like to do?
NanoCDN Storage driver
import hashlib
import urllib.parse
import os
from io import BytesIO, StringIO
import requests
from django.conf import settings
from django.core.exceptions import SuspiciousFileOperation
from django.core.files import File
from import Storage
class NanoCDNFile(File):
def __init__(self, name, storage, mode='rb'):
self.mode = mode = name
self._storage = storage
self._is_read = False
self.file = BytesIO()
self._resp = None
def write(self, content):
raise NotImplemented()
def _read(self):
self._resp = self._storage._read(
b = self._resp.content
if 'b' not in self.mode:
self.file = StringIO(b.decode(self._resp.encoding or 'utf-8'))
self.file = BytesIO(b)
self._is_read = True
return b
def size(self):
if not hasattr(self, '_size'):
if self._is_read:
self._size = self._resp['Content-Length']
return self._size
def read(self, num_bytes=None):
if not self._is_read:
class NanoCDNStorage(Storage):
def __init__(self):
self.base_url = settings.NANOCDN_URL
def _open(self, name, mode='rb'):
return NanoCDNFile(name, self, mode)
def _read(self, name):
resp = requests.get(urllib.parse.urljoin(self.base_url, name), stream=True)
return resp
def _save(self, name, content):
content =
sha1 = hashlib.sha1()
sha1.update(content.encode() if isinstance(content, str) else content)
parts = name.split('/')
if parts[1] in ('pub', 'priv'):
parts = parts[1:]
elif parts[0] not in ('pub', 'priv'):
parts = ['priv'] + parts
name = '/'.join(parts)
if '.' in os.path.basename(name):
bname, ext = os.path.basename(name).rsplit('.', 1)
name = os.path.join(os.path.dirname(name), bname + '.' + sha1.hexdigest()[:14] + '.' + ext)
name = os.path.join(os.path.dirname(name), os.path.basename(name) + '.' + sha1[:14])
resp = requests.put(
urllib.parse.urljoin(self.base_url, os.path.join('upload', name)),
if resp.status_code != 409:
loc = resp.headers['Location']
if loc.startswith('/'):
loc = loc[1:]
return loc
def get_available_name(self, name, max_length=None):
if max_length and len(name) + 15 > max_length:
raise SuspiciousFileOperation(
'Storage can not find an available filename for "%s". '
'Please make sure that the corresponding file field '
'allows sufficient "max_length".' % name
return name
def delete(self, name):
if isinstance(name, NanoCDNFile):
name =
resp = requests.delete(urllib.parse.urljoin(self.base_url, name))
if resp.status_code == 404:
return resp # That is fine
return resp
def exists(self, name):
resp = requests.head(urllib.parse.urljoin(self.base_url, name))
if resp.status_code == 404:
return False
return True
def size(self, name):
resp = requests.head(urllib.parse.urljoin(self.base_url, name))
return resp['Content-Length']
def url(self, name):
return urllib.parse.urljoin(settings.MEDIA_URL, name)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment