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)
- 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 = {
- "s3+private": "private",
- "s3+public-read": "public-read",
- "s3+public-read-write": "public-read-write",
- "s3+aws-exec-read": "aws-exec-read",
- "s3+authenticated-read": "authenticated-read",
- "s3+bucket-owner-read": "bucket-owner-read",
- "s3+bucket-owner-full-control": "bucket-owner-full-control"
- }.get(self.parse.scheme, "private")
- 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)
- scheme = parse.scheme.split("+")[0]
- wrapper = {"s3": AmazonS3}.get(scheme, WebDAV)
- return wrapper(url)
- 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