(self)
Code to initialize the plugin
Code to initialize the plugin
Code to initialize the plugin
def initialize_plugin(self) -> None:
        """
        Code to initialize the plugin
        """
        region_name = self.get_config_value("region_name")
        aws_access_key_id = self.get_config_value("aws_access_key_id")
        aws_secret_access_key = self.get_config_value("aws_secret_access_key")
        endpoint_url = self.get_config_value("endpoint_url")
        signature_version = self.get_config_value("signature_version")
        try:
            mirror_base_path = PureS3Path(self.configuration.get("mirror", "directory"))
        except (configparser.NoOptionError, configparser.NoSectionError) as e:
            logger.error(
                "Mirror directory must be set when using s3 as storage backend"
            )
            raise e
        s3_args = {}
        if endpoint_url:
            s3_args["endpoint_url"] = endpoint_url
        if aws_access_key_id:
            s3_args["aws_access_key_id"] = aws_access_key_id
        if region_name:
            s3_args["region_name"] = region_name
        if aws_secret_access_key:
            s3_args["aws_secret_access_key"] = aws_secret_access_key
        if signature_version:
            s3_args["config"] = Config(signature_version=signature_version)
        resource = boto3.resource("s3", **s3_args)
        register_configuration_parameter(mirror_base_path, resource=resource)
def initialize_plugin ( self ) -> None : region_name = self . get_config_value ( "region_name" ) aws_access_key_id = self . get_config_value ( "aws_access_key_id" ) aws_secret_access_key = self . get_config_value ( "aws_secret_access_key" ) endpoint_url = self . get_config_value ( "endpoint_url" ) signature_version = self . get_config_value ( "signature_version" ) try : mirror_base_path = PureS3Path ( self . configuration . get ( "mirror" , "directory" ) ) except ( configparser . NoOptionError , configparser . NoSectionError ) as e : logger . error ( "Mirror directory must be set when using s3 as storage backend" ) raise e s3_args = { } if endpoint_url : s3_args [ "endpoint_url" ] = endpoint_url if aws_access_key_id : s3_args [ "aws_access_key_id" ] = aws_access_key_id if region_name : s3_args [ "region_name" ] = region_name if aws_secret_access_key : s3_args [ "aws_secret_access_key" ] = aws_secret_access_key if signature_version : s3_args [ "config" ] = Config ( signature_version = signature_version ) resource = boto3 . resource ( "s3" , * * s3_args ) register_configuration_parameter ( mirror_base_path , resource = resource )