Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- # -*- coding: utf-8 -*-
- #
- # Author Ondrej Barta
- # ondrej@ondrej.it
- # Copyright 2017
- import os
- import boto3
- import easywebdav
- import magic
- import tempfile
- from urllib import unquote
- from urlparse import urlparse
- class BaseStorage(object):
- def __init__(self, url):
- self.parse = urlparse(url)
- def get_public_url(self):
- raise NotImplementedError
- def save(self, data):
- raise NotImplementedError
- class AmazonS3(BaseStorage):
- def __init__(self, url):
- super(AmazonS3, self).__init__(url, acl="private")
- self.region = os.environ.get("AWS_REGION", "eu-west-1")
- self.storage_class = os.environ.get("AWS_STORAGE_CLASS", "STANDARD_IA")
- self.username = unquote(self.parse.username)
- self.password = unquote(self.parse.password)
- self.key = self.parse.path.lstrip("/")
- self.acl = acl
- def get_public_url(self, expires_in=3600 * 24):
- s3 = boto3.client(
- "s3",
- aws_access_key_id=self.username,
- aws_secret_access_key=self.password,
- region_name=self.region,
- )
- return s3.generate_presigned_url(
- ClientMethod="get_object",
- Params={
- "Bucket": self.parse.hostname,
- "Key": self.key,
- },
- ExpiresIn=expires_in
- )
- def save(self, data):
- s3 = boto3.resource(
- "s3",
- aws_access_key_id=self.username,
- aws_secret_access_key=self.password,
- region_name=self.region,
- )
- s3.Object(self.parse.hostname, self.key).put(
- Body=data,
- ContentType=get_content_type(data),
- ACL=self.acl,
- StorageClass=self.storage_class
- )
- class WebDAV(BaseStorage):
- def get_public_url(self):
- return self.parse.geturl()
- def save(self, data, verify_ssl=False):
- dav = easywebdav.connect(
- host=self.parse.netloc,
- protocol=self.parse.scheme,
- username=self.parse.username,
- password=self.parse.password,
- verify_ssl=verify_ssl
- )
- dav.upload(data, self.parse.path)
- def storage_driver(url):
- parse = urlparse(url)
- wrapper, params = {
- "s3": (AmazonS3, dict()),
- "s3-public": (AmazonS3, dict(acl="public")),
- "s3-public-read": (AmazonS3, dict(acl="public-read")),
- }.get(scheme, (WebDAV, dict()))
- return wrapper(url, **params)
- def get_content_type(temp_file):
- path = os.path.join(tempfile.gettempdir(), temp_file.name)
- mime = magic.from_file(path, mime=True)
- return mime
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement