全部产品
Search
文档中心

对象存储 OSS:Python快速入门

更新时间:Feb 18, 2025

本文介绍如何快速使用OSS Python SDK完成常见操作,如创建存储空间(Bucket)、上传文件(Object)、下载文件等。

注意事项

  • 关于OSS支持的Region与Endpoint的对应关系,请参见OSS地域和访问域名

  • 本文以从环境变量读取访问凭证为例。更多配置访问凭证的示例,请参见配置访问凭证

  • 本文通过默认示例对Python SDK进行初始化。更多场景示例,请参见初始化

前提条件

配置凭证

  1. 使用RAM用户AccessKey配置环境变量。

    Linux

    1. 在命令行界面执行以下命令来将环境变量设置追加到~/.bashrc 文件中。

      echo "export OSS_ACCESS_KEY_ID='YOUR_ACCESS_KEY_ID'" >> ~/.bashrc
      echo "export OSS_ACCESS_KEY_SECRET='YOUR_ACCESS_KEY_SECRET'" >> ~/.bashrc
      1. 执行以下命令使变更生效。

        source ~/.bashrc
      2. 执行以下命令检查环境变量是否生效。

        echo $OSS_ACCESS_KEY_ID
        echo $OSS_ACCESS_KEY_SECRET

    macOS

    1. 在终端中执行以下命令,查看默认Shell类型。

      echo $SHELL
      1. 根据默认Shell类型进行操作。

        Zsh

        1. 执行以下命令来将环境变量设置追加到 ~/.zshrc 文件中。

          echo "export OSS_ACCESS_KEY_ID='YOUR_ACCESS_KEY_ID'" >> ~/.zshrc
          echo "export OSS_ACCESS_KEY_SECRET='YOUR_ACCESS_KEY_SECRET'" >> ~/.zshrc
        2. 执行以下命令使变更生效。

          source ~/.zshrc
        3. 执行以下命令检查环境变量是否生效。

          echo $OSS_ACCESS_KEY_ID
          echo $OSS_ACCESS_KEY_SECRET

        Bash

        1. 执行以下命令来将环境变量设置追加到 ~/.bash_profile 文件中。

          echo "export OSS_ACCESS_KEY_ID='YOUR_ACCESS_KEY_ID'" >> ~/.bash_profile
          echo "export OSS_ACCESS_KEY_SECRET='YOUR_ACCESS_KEY_SECRET'" >> ~/.bash_profile
        2. 执行以下命令使变更生效。

          source ~/.bash_profile
        3. 执行以下命令检查环境变量是否生效。

          echo $OSS_ACCESS_KEY_ID
          echo $OSS_ACCESS_KEY_SECRET

    Windows

    CMD

    1. 在CMD中运行以下命令。

      setx OSS_ACCESS_KEY_ID "YOUR_ACCESS_KEY_ID"
      setx OSS_ACCESS_KEY_SECRET "YOUR_ACCESS_KEY_SECRET"
      1. 运行以下命令,检查环境变量是否生效。

        echo %OSS_ACCESS_KEY_ID%
        echo %OSS_ACCESS_KEY_SECRET%

    PowerShell

    1. 在PowerShell中运行以下命令。

      [Environment]::SetEnvironmentVariable("OSS_ACCESS_KEY_ID", "YOUR_ACCESS_KEY_ID", [EnvironmentVariableTarget]::User)
      [Environment]::SetEnvironmentVariable("OSS_ACCESS_KEY_SECRET", "YOUR_ACCESS_KEY_SECRET", [EnvironmentVariableTarget]::User)
      1. 运行以下命令,检查环境变量是否生效。

        [Environment]::GetEnvironmentVariable("OSS_ACCESS_KEY_ID", [EnvironmentVariableTarget]::User)
        [Environment]::GetEnvironmentVariable("OSS_ACCESS_KEY_SECRET", [EnvironmentVariableTarget]::User)

  2. 参考上述方式修改系统环境变量后,请重启或刷新您的编译运行环境,包括IDE、命令行界面、其他桌面应用程序及后台服务,以确保最新的系统环境变量成功加载。

安装SDK

  1. 已安装合适版本的Python运行环境,通过以下命令查看Python版本。

    OSS Python SDK适用于Python 2.6、2.7、3.3、3.4、3.5、3.6、3.7、3.8及以上版本。

    python -version

    如果当前计算环境没有安装Python运行环境,请下载安装Python

  2. 安装OSS Python SDK。

    执行以下命令安装SDK。

    pip install oss2

运行示例

运行以下代码示例以体验OSS的完整使用流程:创建一个Bucket、上传文件、下载文件、列举文件以及删除文件和Bucket。

# -*- coding: utf-8 -*-
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider
from itertools import islice
import os
import logging
import time
import random

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# 检查环境变量是否已设置
required_env_vars = ['OSS_ACCESS_KEY_ID', 'OSS_ACCESS_KEY_SECRET']
for var in required_env_vars:
    if var not in os.environ:
        logging.error(f"Environment variable {var} is not set.")
        exit(1)

# 从环境变量中获取访问凭证
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# 设置Endpoint和Region
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"
region = "cn-hangzhou"

def generate_unique_bucket_name():
    # 获取当前时间戳
    timestamp = int(time.time())
    # 生成0到9999之间的随机数
    random_number = random.randint(0, 9999)
    # 构建唯一的Bucket名称
    bucket_name = f"demo-{timestamp}-{random_number}"
    return bucket_name

# 生成唯一的Bucket名称
bucket_name = generate_unique_bucket_name()
bucket = oss2.Bucket(auth, endpoint, bucket_name, region=region)

def create_bucket(bucket):
    try:
        bucket.create_bucket(oss2.models.BUCKET_ACL_PRIVATE)
        logging.info("Bucket created successfully")
    except oss2.exceptions.OssError as e:
        logging.error(f"Failed to create bucket: {e}")

def upload_file(bucket, object_name, data):
    try:
        result = bucket.put_object(object_name, data)
        logging.info(f"File uploaded successfully, status code: {result.status}")
    except oss2.exceptions.OssError as e:
        logging.error(f"Failed to upload file: {e}")

def download_file(bucket, object_name):
    try:
        file_obj = bucket.get_object(object_name)
        content = file_obj.read().decode('utf-8')
        logging.info("File content:")
        logging.info(content)
        return content
    except oss2.exceptions.OssError as e:
        logging.error(f"Failed to download file: {e}")

def list_objects(bucket):
    try:
        objects = list(islice(oss2.ObjectIterator(bucket), 10))
        for obj in objects:
            logging.info(obj.key)
    except oss2.exceptions.OssError as e:
        logging.error(f"Failed to list objects: {e}")

def delete_objects(bucket):
    try:
        objects = list(islice(oss2.ObjectIterator(bucket), 100))
        if objects:
            for obj in objects:
                bucket.delete_object(obj.key)
                logging.info(f"Deleted object: {obj.key}")
        else:
            logging.info("No objects to delete")
    except oss2.exceptions.OssError as e:
        logging.error(f"Failed to delete objects: {e}")

def delete_bucket(bucket):
    try:
        bucket.delete_bucket()
        logging.info("Bucket deleted successfully")
    except oss2.exceptions.OssError as e:
        logging.error(f"Failed to delete bucket: {e}")

# 主流程
if __name__ == '__main__':
    # 1. 创建Bucket
    create_bucket(bucket)
    # 2. 上传文件
    upload_file(bucket, 'test-string-file', b'Hello OSS, this is a test string.')
    # 3. 下载文件
    download_file(bucket, 'test-string-file')
    # 4. 列出Bucket中的对象
    list_objects(bucket)
    # 5. 删除Bucket中的对象
    delete_objects(bucket)
    # 6. 删除Bucket
    delete_bucket(bucket)
    

常见问题

当使用SDK时遇到报错AccessDenied该如何进行排查?

当您遇到AccessDenied错误时通常是因为没有相应的访问权限。以下是解决此问题的步骤:

  1. 确认AccessKey ID和AccessKey Secret:请确保您使用了正确的AccessKey ID和AccessKey Secret。详情请参见创建AccessKey

  2. 检查RAM用户权限:确认RAM用户是否拥有Bucket或Object相关操作权限。具体步骤,请参见为RAM用户授权

  3. 检查Bucket Policy:如果错误信息中提到“Access denied by bucket policy”,则表明是受到Bucket Policy的策略拒绝了。详情请参考Bucket Policy

  4. 更多错误类型的查询,请您参阅EC错误码,例如关于访问控制方面的常见报错,可参见03-ACCESS_CONTROL