wsstorage.py 文件源码

def __create_signed_url_for_resource(
            self,
            verb="GET",
            bucket=None,
            file_key=None,
            duration=config.storage_signed_url_duration,
            creds_file_path=config.gcp_creds_file_path,
    ):
        """
        Create and return a signed URL for retrieving the specified file from GCP.
        :param verb: The HTTP verb for the signed request.
        :param bucket: The bucket where the file resides.
        :param file_key: The key where the file resides within the bucket.
        :param duration: The amount of time in seconds that the URL should be valid for.
        :param creds_file_path: The local file path to where the GCP credentials to use to sign
        the URL reside.
        :return: A signed URL that can be used to retrieve the referenced file's contents.
        """
        to_sign, expires_epoch = self.__get_signing_content_for_resource(
            verb=verb,
            bucket=bucket,
            file_key=file_key,
            duration=duration,
        )
        creds = ServiceAccountCredentials.from_json_keyfile_name(creds_file_path)
        client_id = creds.service_account_email
        signed_blob = creds.sign_blob(to_sign)[1]
        encoded_sig = b64encode(signed_blob).replace("+", "%2B").replace("/", "%2F")
        resource_url = "%s%s/%s" % (
            self.base_url,
            bucket,
            file_key,
        )
        return "%s?GoogleAccessId=%s&Expires=%s&Signature=%s" % (
            resource_url,
            client_id,
            expires_epoch,
            encoded_sig,
        )