Source code for flask_s3

import gzip
import hashlib
import json
import logging
import os
import re

    from cStringIO import StringIO
except ImportError:
    from io import StringIO
import mimetypes
from collections import defaultdict

import boto3
import boto3.exceptions
from botocore.exceptions import ClientError
from flask import current_app
from flask import url_for as flask_url_for
import six

logger = logging.getLogger('flask_s3')

# Mapping for Header names to S3 parameters
header_mapping = {
    'cache-control': 'CacheControl',
    'content-disposition': 'ContentDisposition',
    'content-encoding': 'ContentEncoding',
    'content-language': 'ContentLanguage',
    'content-length': 'ContentLength',
    'content-md5': 'ContentMD5',
    'content-type': 'ContentType',
    'expires': 'Expires',

                    'FLASKS3_ACTIVE': True,
                    'FLASKS3_DEBUG': False,
                    'FLASKS3_BUCKET_DOMAIN': '',
                    'FLASKS3_CDN_DOMAIN': '',
                    'FLASKS3_USE_CACHE_CONTROL': False,
                    'FLASKS3_HEADERS': {},
                    'FLASKS3_FILEPATH_HEADERS': {},
                    'FLASKS3_ONLY_MODIFIED': False,
                    'FLASKS3_URL_STYLE': 'host',
                    'FLASKS3_ENDPOINT_URL': None,
                    'FLASKS3_GZIP': False,
                    'FLASKS3_GZIP_ONLY_EXTS': [],
                    'FLASKS3_FORCE_MIMETYPE': False,
                    'FLASKS3_PREFIX': ''}

__version__ = (0, 3, 2)

def _get_statics_prefix(app):
    Get the complete prefix that should be used by static files.
    upload_prefix = app.config.get('FLASKS3_PREFIX', '')
    return '/%s' % upload_prefix.lstrip('/').rstrip('/')

def split_metadata_params(headers):
    Given a dict of headers for s3, seperates those that are boto3
    parameters and those that must be metadata

    params = {}
    metadata = {}
    for header_name in headers:
        if header_name.lower() in header_mapping:
            params[header_mapping[header_name.lower()]] = headers[header_name]
            metadata[header_name] = headers[header_name]
    return metadata, params

def merge_two_dicts(x, y):
    """Given two dicts, merge them into a new dict as a shallow copy."""
    z = x.copy()
    return z

def hash_file(filename):
    Generate a hash for the contents of a file
    hasher = hashlib.sha1()
    with open(filename, 'rb') as f:
        buf =
        while len(buf) > 0:
            buf =

    return hasher.hexdigest()

def _get_bucket_name(**values):
    Generates the bucket name for url_for.
    app = current_app
    # manage other special values, all have no meaning for static urls
    values.pop('_external', False)  # external has no meaning here
    values.pop('_anchor', None)  # anchor as well
    values.pop('_method', None)  # method too

    url_style = get_setting('FLASKS3_URL_STYLE', app)
    if url_style == 'host':
        url_format = '{bucket_name}.{bucket_domain}'
    elif url_style == 'path':
        url_format = '{bucket_domain}/{bucket_name}'
        raise ValueError('Invalid S3 URL style: "{}"'.format(url_style))

    if get_setting('FLASKS3_CDN_DOMAIN', app):
        bucket_path = '{}'.format(get_setting('FLASKS3_CDN_DOMAIN', app))

        bucket_path = url_format.format(
            bucket_name=get_setting('FLASKS3_BUCKET_NAME', app),
            bucket_domain=get_setting('FLASKS3_BUCKET_DOMAIN', app),

    bucket_path += _get_statics_prefix(app).rstrip('/')

    return bucket_path, values

[docs]def url_for(endpoint, **values): """ Generates a URL to the given endpoint. If the endpoint is for a static resource then an Amazon S3 URL is generated, otherwise the call is passed on to `flask.url_for`. Because this function is set as a jinja environment variable when `FlaskS3.init_app` is invoked, this function replaces `flask.url_for` in templates automatically. It is unlikely that this function will need to be directly called from within your application code, unless you need to refer to static assets outside of your templates. """ app = current_app if app.config.get('TESTING', False) and not app.config.get('FLASKS3_OVERRIDE_TESTING', True): return flask_url_for(endpoint, **values) if 'FLASKS3_BUCKET_NAME' not in app.config: raise ValueError("FLASKS3_BUCKET_NAME not found in app configuration.") if endpoint == 'static' or endpoint.endswith('.static'): scheme = 'https' if not app.config.get("FLASKS3_USE_HTTPS", True): scheme = 'http' # allow per url override for scheme scheme = values.pop('_scheme', scheme) bucket_path, values = _get_bucket_name(**values) urls = app.url_map.bind(bucket_path, url_scheme=scheme) built =, values=values, force_external=True) return built return flask_url_for(endpoint, **values)
def _bp_static_url(blueprint): """ builds the absolute url path for a blueprint's static folder """ u = six.u('%s%s' % (blueprint.url_prefix or '', blueprint.static_url_path or '')) return u def _gather_files(app, hidden, filepath_filter_regex=None): """ Gets all files in static folders and returns in dict.""" dirs = [(six.text_type(app.static_folder), app.static_url_path)] if hasattr(app, 'blueprints'): blueprints = app.blueprints.values() bp_details = lambda x: (x.static_folder, _bp_static_url(x)) dirs.extend([bp_details(x) for x in blueprints if x.static_folder]) valid_files = defaultdict(list) for static_folder, static_url_loc in dirs: if not os.path.isdir(static_folder): logger.warning("WARNING - [%s does not exist]" % static_folder) else: logger.debug("Checking static folder: %s" % static_folder) for root, _, files in os.walk(static_folder): relative_folder = re.sub(r'^/', '', root.replace(static_folder, '')) files = [os.path.join(root, x) for x in files if ( (hidden or x[0] != '.') and # Skip this file if the filter regex is # defined, and this file's path is a # negative match. (filepath_filter_regex == None or filepath_filter_regex, os.path.join(relative_folder, x))))] if files: valid_files[(static_folder, static_url_loc)].extend(files) return valid_files def _path_to_relative_url(path): """ Converts a folder and filename into a ralative url path """ return os.path.splitdrive(path)[1].replace('\\', '/') def _static_folder_path(static_url, static_folder, static_asset): """ Returns a path to a file based on the static folder, and not on the filesystem holding the file. Returns a path relative to static_url for static_asset """ # first get the asset path relative to the static folder. # static_asset is not simply a filename because it could be # sub-directory then file etc. if not static_asset.startswith(static_folder): raise ValueError("%s static asset must be under %s static folder" % (static_asset, static_folder)) rel_asset = static_asset[len(static_folder):] # Now bolt the static url path and the relative asset location together return '%s/%s' % (static_url.rstrip('/'), rel_asset.lstrip('/')) def _write_files(s3, app, static_url_loc, static_folder, files, bucket, ex_keys=None, hashes=None): """ Writes all the files inside a static folder to S3. """ should_gzip = app.config.get('FLASKS3_GZIP') add_mime = app.config.get('FLASKS3_FORCE_MIMETYPE') gzip_include_only = app.config.get('FLASKS3_GZIP_ONLY_EXTS') new_hashes = [] static_folder_rel = _path_to_relative_url(static_folder) for file_path in files: per_file_should_gzip = should_gzip asset_loc = _path_to_relative_url(file_path) full_key_name = _static_folder_path(static_url_loc, static_folder_rel, asset_loc) key_name = full_key_name.lstrip("/") logger.debug("Uploading {} to {} as {}".format(file_path, bucket, key_name)) exclude = False if app.config.get('FLASKS3_ONLY_MODIFIED', False): file_hash = hash_file(file_path) new_hashes.append((full_key_name, file_hash)) if hashes and hashes.get(full_key_name, None) == file_hash: exclude = True if ex_keys and full_key_name in ex_keys or exclude: logger.debug("%s excluded from upload" % key_name) else: h = {} # Set more custom headers if the filepath matches certain # configured regular expressions. filepath_headers = app.config.get('FLASKS3_FILEPATH_HEADERS') if filepath_headers: for filepath_regex, headers in six.iteritems(filepath_headers): if, file_path): for header, value in six.iteritems(headers): h[header] = value # check for extension, only if there are extensions provided if per_file_should_gzip and gzip_include_only: if os.path.splitext(file_path)[1] not in gzip_include_only: per_file_should_gzip = False if per_file_should_gzip: h["content-encoding"] = "gzip" if (add_mime or per_file_should_gzip) and "content-type" not in h: # When we use GZIP we have to explicitly set the content type # or if the mime flag is True (mimetype, encoding) = mimetypes.guess_type(file_path, False) if mimetype: h["content-type"] = mimetype else: logger.warn("Unable to detect mimetype for %s" % file_path) file_mode = 'rb' if six.PY3 else 'r' with open(file_path, file_mode) as fp: merged_dicts = merge_two_dicts(get_setting('FLASKS3_HEADERS', app), h) metadata, params = split_metadata_params(merged_dicts) if per_file_should_gzip: compressed = six.BytesIO() z = gzip.GzipFile(os.path.basename(file_path), 'wb', 9, compressed) z.write( z.close() data = compressed.getvalue() else: data = s3.put_object(Bucket=bucket, Key=key_name, Body=data, ACL="public-read", Metadata=metadata, **params) return new_hashes def _upload_files(s3, app, files_, bucket, hashes=None): new_hashes = [] prefix = _get_statics_prefix(app) for (static_folder, static_url), names in six.iteritems(files_): static_upload_url = '%s/%s' % (prefix.rstrip('/'), static_url.lstrip('/')) new_hashes.extend(_write_files(s3, app, static_upload_url, static_folder, names, bucket, hashes=hashes)) return new_hashes def get_setting(name, app=None): """ Returns the value for `name` settings (looks into `app` config, and into DEFAULT_SETTINGS). Returns None if not set. :param name: (str) name of a setting (e.g. FLASKS3_URL_STYLE) :param app: Flask app instance :return: setting value or None """ default_value = DEFAULT_SETTINGS.get(name, None) return app.config.get(name, default_value) if app else default_value
[docs]def create_all(app, user=None, password=None, bucket_name=None, location=None, include_hidden=False, filepath_filter_regex=None, put_bucket_acl=True): """ Uploads of the static assets associated with a Flask application to Amazon S3. All static assets are identified on the local filesystem, including any static assets associated with *registered* blueprints. In turn, each asset is uploaded to the bucket described by `bucket_name`. If the bucket does not exist then it is created. Flask-S3 creates the same relative static asset folder structure on S3 as can be found within your Flask application. Many of the optional arguments to `create_all` can be specified instead in your application's configuration using the Flask-S3 `configuration`_ variables. :param app: a :class:`flask.Flask` application object. :param user: an AWS Access Key ID. You can find this key in the Security Credentials section of your AWS account. :type user: `basestring` or None :param password: an AWS Secret Access Key. You can find this key in the Security Credentials section of your AWS account. :type password: `basestring` or None :param bucket_name: the name of the bucket you wish to server your static assets from. **Note**: while a valid character, it is recommended that you do not include periods in bucket_name if you wish to serve over HTTPS. See Amazon's `bucket restrictions`_ for more details. :type bucket_name: `basestring` or None :param location: the AWS region to host the bucket in; an empty string indicates the default region should be used, which is the US Standard region. Possible location values include: `'DEFAULT'`, `'EU'`, `'us-east-1'`, `'us-west-1'`, `'us-west-2'`, `'ap-south-1'`, `'ap-northeast-2'`, `'ap-southeast-1'`, `'ap-southeast-2'`, `'ap-northeast-1'`, `'eu-central-1'`, `'eu-west-1'`, `'sa-east-1'` :type location: `basestring` or None :param include_hidden: by default Flask-S3 will not upload hidden files. Set this to true to force the upload of hidden files. :type include_hidden: `bool` :param filepath_filter_regex: if specified, then the upload of static assets is limited to only those files whose relative path matches this regular expression string. For example, to only upload files within the 'css' directory of your app's static store, set to r'^css'. :type filepath_filter_regex: `basestring` or None :param put_bucket_acl: by default Flask-S3 will set the bucket ACL to public. Set this to false to leave the policy unchanged. :type put_bucket_acl: `bool` .. _bucket restrictions:\ /latest/dev/BucketRestrictions.html """ user = user or app.config.get('AWS_ACCESS_KEY_ID') password = password or app.config.get('AWS_SECRET_ACCESS_KEY') bucket_name = bucket_name or app.config.get('FLASKS3_BUCKET_NAME') if not bucket_name: raise ValueError("No bucket name provided.") location = location or app.config.get('FLASKS3_REGION') endpoint_url = app.config.get('FLASKS3_ENDPOINT_URL') # build list of static files all_files = _gather_files(app, include_hidden, filepath_filter_regex=filepath_filter_regex) logger.debug("All valid files: %s" % all_files) # connect to s3 s3 = boto3.client("s3", endpoint_url=endpoint_url, region_name=location or None, aws_access_key_id=user, aws_secret_access_key=password) # get_or_create bucket try: s3.head_bucket(Bucket=bucket_name) except ClientError as e: if int(e.response['Error']['Code']) == 404: # Create the bucket bucket = s3.create_bucket(Bucket=bucket_name) else: raise if put_bucket_acl: s3.put_bucket_acl(Bucket=bucket_name, ACL='public-read') if get_setting('FLASKS3_ONLY_MODIFIED', app): try: hashes_object = s3.get_object(Bucket=bucket_name, Key='.file-hashes') hashes = json.loads(str(hashes_object['Body'].read().decode())) except ClientError as e: logger.warn("No file hashes found: %s" % e) hashes = None new_hashes = _upload_files(s3, app, all_files, bucket_name, hashes=hashes) try: s3.put_object(Bucket=bucket_name, Key='.file-hashes', Body=json.dumps(dict(new_hashes)), ACL='private') except boto3.exceptions.S3UploadFailedError as e: logger.warn("Unable to upload file hashes: %s" % e) else: _upload_files(s3, app, all_files, bucket_name)
[docs]class FlaskS3(object): """ The FlaskS3 object allows your application to use Flask-S3. When initialising a FlaskS3 object you may optionally provide your :class:`flask.Flask` application object if it is ready. Otherwise, you may provide it later by using the :meth:`init_app` method. :param app: optional :class:`flask.Flask` application object :type app: :class:`flask.Flask` or None """ def __init__(self, app=None): if app is not None: self.init_app(app)
[docs] def init_app(self, app): """ An alternative way to pass your :class:`flask.Flask` application object to Flask-S3. :meth:`init_app` also takes care of some default `settings`_. :param app: the :class:`flask.Flask` application object. """ for k, v in DEFAULT_SETTINGS.items(): app.config.setdefault(k, v) if app.debug and not get_setting('FLASKS3_DEBUG', app): app.config['FLASKS3_ACTIVE'] = False if get_setting('FLASKS3_ACTIVE', app): app.jinja_env.globals['url_for'] = url_for if get_setting('FLASKS3_USE_CACHE_CONTROL', app) and app.config.get('FLASKS3_CACHE_CONTROL'): cache_control_header = get_setting('FLASKS3_CACHE_CONTROL', app) app.config['FLASKS3_HEADERS']['Cache-Control'] = cache_control_header