Django Admin 上傳文件到七牛雲的示例代碼

中文圈關於Django Admin 上傳文件到七牛雲的資料和函數庫已經是2年前的了，國外的則都是關於AWS S3、Azure Storage一些國外的服務的。我根據Django的文檔裡提到的存儲系統來實現上傳文件到七牛雲的簡單功能。

在Django Admin的表單是根據數據模型生成的，其中文件上傳由FileField和繼承FileField的ImageField來決定的，文件上傳到靜態文件目錄，數據庫保存相對路徑。實現上傳文件到七牛雲我們是根據FileField的storage參數來實現的。

models.ImageField(storage=MyStorage())

storage參數需要傳入一個實現抽象類Storage的對象。

下面引用文檔的原句

你自定義的存儲系統必須為Django.core.files.storage.Storage的一個子類:

  from django.conf import settings  from django.core.files.storage import Storage    class MyStorage(Storage):    def __init__(self, option=None):      if not option:        option = settings.CUSTOM_STORAGE_OPTIONS      ...

然後重寫全部的抽象方法，具體的可以去官網文檔看。我們這裡先不繼承Storage。我在Django的核心裡找到了已經繼承了的FileSystemStorage，我想只要繼承Django標準的文件管理類

FileSystemStorage，重載_save部分變成上傳到七牛雲不就好了。說幹就幹。

  class MyStorage(FileSystemStorage):    def _save(self, name, content):      # 延續原方法的寫法      filename = name.replace('', '/')      # 將文件傳入封裝好的對象裡      q = Qiniu()      q.upload_stream(filename, content.file.getvalue())      return filename

我簡單寫一下封裝的Qiniu上傳的類

  class Qiniu():    def __init__(self):      self.access_key = settings.QINIU_ACCESS_KEY      self.secret_key = settings.QINIU_SECRET_KEY      # 要上傳的空間      self.bucket_name = settings.QINIU_BUCKET_NAME      # 構建鑑權對象      self.auth = Auth(self.access_key, self.secret_key)          def get_token(self, key):      """        :param key: 文件名      :return: 上傳令牌      """      policy = {        'scope': settings.QINIU_BUCKET_NAME,        'mimeLimit': 'image/jpeg;image/png',        'deadline': 3600      }      # 3600為token過期時間，秒為單位。3600等於一小時      token = self.auth.upload_token(self.bucket_name, key, 3600, policy)      return token          def upload_stream(self, filename, stream_data):      """        :param filename: 文件名      :param stream_data: 二進制數據      :return: 無      """      # 上傳後保存的文件名      key = filename      # 生成上傳 Token，可以指定過期時間等      token = self.auth.upload_token(self.bucket_name, key, 3600)      # 要上傳文件的本地路徑      # localfile = file_path      ret, info = put_data(up_token=token, key=key, data=stream_data)      assert ret['key'] == key      # assert ret['hash'] == etag_stream(stream_data)