Source code for invenio_s3.ext

# -*- coding: utf-8 -*-
#
# Copyright (C) 2018, 2019 Esteban J. G. Gabancho.
#
# Invenio-S3 is free software; you can redistribute it and/or modify it
# under the terms of the MIT License; see LICENSE file for more details.
"""S3 file storage support for Invenio."""

from __future__ import absolute_import, print_function

import warnings

import boto3
from flask import current_app
from werkzeug.utils import cached_property

from . import config


class InvenioS3(object):
    """Invenio-S3 extension."""

    def __init__(self, app=None):
        """Extension initialization."""
        if app:
            self.init_app(app)

    @cached_property
    def init_s3fs_info(self):
        """Gather all the information needed to start the S3FSFileSystem."""
        if "S3_ACCCESS_KEY_ID" in current_app.config:
            current_app.config["S3_ACCESS_KEY_ID"] = current_app.config[
                "S3_ACCCESS_KEY_ID"
            ]
            warnings.warn(
                "Key S3_ACCCESS_KEY_ID contained a typo and has been "
                "corrected to S3_ACCESS_KEY_ID, support for the "
                "flawed version will be removed.",
                DeprecationWarning,
            )

        if "S3_SECRECT_ACCESS_KEY" in current_app.config:
            current_app.config["S3_SECRET_ACCESS_KEY"] = current_app.config[
                "S3_SECRECT_ACCESS_KEY"
            ]
            warnings.warn(
                "Key S3_SECRECT_ACCESS_KEY contained a typo and has been "
                "corrected to S3_SECRET_ACCESS_KEY, support for the "
                "flawed version will be removed.",
                DeprecationWarning,
            )

        info = dict(
            key=current_app.config.get("S3_ACCESS_KEY_ID", ""),
            secret=current_app.config.get("S3_SECRET_ACCESS_KEY", ""),
            client_kwargs={},
            config_kwargs={
                "s3": {
                    "addressing_style": "path",
                },
                "signature_version": current_app.config.get(
                    "S3_SIGNATURE_VERSION", "s3v4"
                ),
            },
        )

        s3_endpoint = current_app.config.get("S3_ENDPOINT_URL", None)
        if s3_endpoint:
            info["client_kwargs"]["endpoint_url"] = s3_endpoint

        region_name = current_app.config.get("S3_REGION_NAME", None)
        if region_name:
            info["client_kwargs"]["region_name"] = region_name

        return info

    def init_app(self, app):
        """Flask application initialization."""
        self.init_config(app)
        app.extensions["invenio-s3"] = self

    def init_config(self, app):
        """Initialize configuration."""
        for k in dir(config):
            if k.startswith("S3_"):
                app.config.setdefault(k, getattr(config, k))