-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcloud_backends.py
More file actions
57 lines (48 loc) · 2.13 KB
/
cloud_backends.py
File metadata and controls
57 lines (48 loc) · 2.13 KB
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
44
45
46
47
48
49
50
51
52
53
54
55
56
57
from abc import ABC, abstractmethod
from typing import List
import dropbox
import cloudinary
import cloudinary.uploader
# local imports
from config import (
CLOUDINARY_CLOUD_NAME, CLOUDINARY_API_KEY, CLOUDINARY_API_SECRET, DROPBOX_API_KEY
)
class CloudBackend(ABC):
@abstractmethod
def upload(self, file_path: str, key: str) -> str:
""" upload local file to remote; returns accessible URL """
pass
class CloudinaryBackend(CloudBackend):
def __init__(self):
#& added this check to the config but it may be better to keep it here - depends on if I ever want to use this without any cloud uploads
# if not (CLOUDINARY_CLOUD_NAME and CLOUDINARY_API_KEY and CLOUDINARY_API_SECRET):
# raise RuntimeError('Missing Cloudinary credentials')
cloudinary.config(
cloud_name=CLOUDINARY_CLOUD_NAME,
api_key=CLOUDINARY_API_KEY,
api_secret=CLOUDINARY_API_SECRET,
secure=True
)
def upload(self, file_path: str, key: str) -> str:
result: dict = cloudinary.uploader.upload(
file_path,
resource_type='raw',
public_id=key
)
return result.get('secure_url')
class DropboxBackend(CloudBackend):
def __init__(self):
if not DROPBOX_API_KEY:
raise RuntimeError('Missing Dropbox API key')
self.dbx = dropbox.Dropbox(DROPBOX_API_KEY)
raise FutureWarning('Dropbox API is deprecated and may not work in the future. Consider using Cloudinary or another service.')
def upload(self, file_path: str, key: str) -> str:
dbx_path = f"/{key}" # Dropbox path must start with '/'
with open(file_path, 'rb') as f:
self.dbx.files_upload(f.read(), dbx_path) # mode=dropbox.files.WriteMode.overwrite)
# return a shareable link
self.dbx.sharing_create_shared_link_with_settings(dbx_path)
links: List[str] = self.dbx.sharing_get_shared_links(dbx_path)
if not links:
raise RuntimeError('Failed to create shared link for Dropbox file')
return str(links[-1])