Repository: spotify/luigi
File: luigi/contrib/redshift.py (Python)

Code before:
    @property
    def aws_account_id(self):
        """
        Override to return the account id.
        """
        return None

    @property
    def aws_arn_role_name(self):
        """
        Override to return the arn role name.
        """
        return None

    @property
    def aws_session_token(self):
        """
        Override to return the session token.
        """
        return None

    def _get_s3_configuration_attribute(self, attribute):
        config = luigi.configuration.get_config()

        try:
            value = config.get('s3', attribute)
        except NoSectionError:
            value = None

        if value is None or value == '':
            value = os.environ.get(attribute.upper(), None)

        return value

    def _credentials(self):
        """
        Return a credentials string for the provided task. If no valid
        credentials are set, raise a NotImplementedError.
        """

        if self.aws_account_id and self.aws_arn_role_name:
            return 'aws_iam_role=arn:aws:iam::{id}:role/{role}'.format(
                id=self.aws_account_id,
                role=self.aws_arn_role_name
            )
        elif self.aws_access_key_id and self.aws_secret_access_key:
            return 'aws_access_key_id={key};aws_secret_access_key={secret}{opt}'.format(
                key=self.aws_access_key_id,
                secret=self.aws_secret_access_key,
                opt=';token={}'.format(self.aws_session_token) if self.aws_session_token else ''
            )

Review comment (suggestion):
Since these are general AWS creds, I'm not sure it makes sense to restrict their config location to `s3`, particularly if they're being used here for Redshift.

Code after:
        """
        return self._get_configuration_attribute('aws_secret_access_key')

    @property
    def aws_account_id(self):
        """
        Override to return the account id.
        """
        return self._get_configuration_attribute('aws_account_id')

    @property
    def aws_arn_role_name(self):
        """
        Override to return the arn role name.
        """
        return self._get_configuration_attribute('aws_arn_role_name')

    @property
    def aws_session_token(self):
        """
        Override to return the session token.
        """
        return self._get_configuration_attribute('aws_session_token')

    def _get_configuration_attribute(self, attribute):
        config = luigi.configuration.get_config()

        value = config.get(self.configuration_section, attribute, default=None)

        if not value:
            value = os.environ.get(attribute.upper(), None)

        return value

    def _credentials(self):
        """
        Return a credential string for the provided task. If no valid
        credentials are set, raise a NotImplementedError.
        """

        if self.aws_account_id and self.aws_arn_role_name:
            return 'aws_iam_role=arn:aws:iam::{id}:role/{role}'.format(
                id=self.aws_account_id,
                role=self.aws_arn_role_name
            )
        elif self.aws_access_key_id and self.aws_secret_access_key:
            return 'aws_access_key_id={key};aws_secret_access_key={secret}{opt}'.format(
                key=self.aws_access_key_id,
                secret=self.aws_secret_access_key,
                opt=';token={}'.format(self.aws_session_token) if self.aws_session_token else ''
            )