接手过不少上传需求，最烦的就是测试环境一切正常，一到生产就各种权限、超时、大文件上传失败。尤其是对接OSS这类对象存储，很多人第一反应是去控制台点点点、看看贴图教程，但实际写代码时反而容易懵。这篇文章不谈花哨的截图，直接给你一套纯代码跑通的方案，围绕Python操作OSS上传这个核心场景，把环境准备、基础上传、大文件分片、工程化封装到线上问题排查完整过一遍。无论你是刚开始接触Python上传的小白，还是已经被线上问题折腾过的老手，照着敲都能落地。
1. OSS到底是什么，为什么大家都在折腾它
1.1 对象存储和传统文件存储的区别
先明确一下概念。OSS（Object Storage Service）是对象存储服务的统称，阿里云叫OSS，腾讯云叫COS，AWS叫S3，各家名字不同，底层逻辑高度一致：把文件当作对象，存到桶（Bucket）里，通过URL访问。
传统方式是把文件塞到服务器本地磁盘，比如Linux的/data/uploads/目录，配个Nginx指向它。早期小项目这么干没问题，但文件一多，磁盘满了、备份麻烦、多个应用服务器之间文件不同步、迁移成本高，痛点会越来越明显。对象存储的核心优势是：存储空间近乎无限、自带容灾冗余、按量付费、通过HTTP接口读写，业务服务器只存路径不存文件本体，天然适合分布式架构。
用生活类比来说，传统文件存储像是你自己家修了个仓库，东西放满了得再盖一间，搬家时全得手动搬。对象存储像是租了专业仓储公司的仓位，你只管把货送过去，取货时凭凭证取，仓位无限大，公司负责安保、防火、备份。
1.2 搞清楚你的上传场景再选方案
在写代码之前，先回答三个问题，不同答案对应完全不同的实现方式。
第一个问题：文件产生在哪端？如果文件在服务端（比如后端收到的上传请求、爬虫抓取的图片、运维上传的备份包），直接用Python SDK在服务端上传。如果文件在浏览器或手机端（用户上传头像、视频），更合理的做法是后端生成一个临时上传凭证（STS或签名URL），前端直传OSS，文件不经过你的业务服务器。很多人一开始就把这两者混在一起，导致服务器带宽被上传流量打满。
第二个问题：文件多大？小于100MB，一次性上传就行。大几百MB甚至GB级别的文件，必须走分片上传（Multipart Upload），否则一个网络抖动就可能让整次上传废掉。对时延敏感的小文件走简单上传接口，对超大文件走断点续传，这个选型后面会展开讲。
第三个问题：是否需要后续处理？比如图片压缩、视频转码、内容审核，如果你选了带数据处理能力的OSS，上传完成后可以自动触发；如果只是存原始文件，那就怎么简单怎么来。
2. 环境准备：从零搭建Python上传OSS的开发环境
2.1 Python环境安装与检查
写Python代码第一步是确保本机有可用的Python解释器。很多入门者卡在环境上，不是代码问题，是环境没弄好。
Windows用户去Python官网下载安装包时，安装过程中务必勾选“Add Python to PATH”。这一步不勾，后面在命令行里敲python会提示找不到命令。macOS用户建议用Homebrew安装：brew install python@3.11。Linux用户多数发行版自带Python 3，检查一下版本就行。
装完验证一下：
python --version # 或者 python3 --version我建议直接用VSCode或PyCharm建项目，VSCode需要手动装Python扩展，然后选解释器。这一套不展开太多，记住一个关键点：Python版本不低于3.8就行，太高版本反而要注意某些依赖兼容性。我自己用的3.10，跑oss2没踩过坑。
2.2 创建项目并安装oss2
OSS的Python SDK，阿里云官方提供的包叫oss2，通过pip直接装：
pip install oss2国内网络环境建议用镜像源加速：
pip install oss2 -i https://pypi.tuna.tsinghua.edu.cn/simple安装完成后，在项目代码里导入验证：
import oss2 print(oss2.__version__)如果输出了版本号，SDK就装好了。这个包依赖requests、crcmod等底层库，pip会自动处理。我遇到过Windows下crcmod装不上的情况，一般升级pip或装crc32c能解决。
2.3 初始化认证信息，先把这几种“钥匙”搞明白
OSS的访问凭证主要有三类：AccessKey（长期密钥）、STS临时凭证（短期、可限定权限）、RAM子账号AccessKey（权限可控的长期密钥）。个人学习直接用主账号AccessKey就行，但生产环境必须用RAM子账号，并且权限要尽量小。
创建RAM子账号的路径：阿里云控制台 -> RAM访问控制 -> 用户 -> 创建用户。创建时勾选“OpenAPI调用访问”，会生成AccessKey ID和AccessKey Secret，记下来，这东西只在创建时完整展示一次。给子账号授权时，比如只允许操作某个Bucket，就用自定义权限策略：
{ "Version": "1", "Statement": [ { "Effect": "Allow", "Action": "oss:PutObject", "Resource": "acs:oss:*:*:your-bucket/*" } ] }然后初始化客户端：
import oss2 access_key_id = "你的AccessKey ID" access_key_secret = "你的AccessKey Secret" endpoint = "oss-cn-hangzhou.aliyuncs.com" # 对应你的Bucket所在地域 bucket_name = "your-bucket" auth = oss2.Auth(access_key_id, access_key_secret) bucket = oss2.Bucket(auth, endpoint, bucket_name)这里最容易踩的坑是endpoint配错。Bucket是杭州的，endpoint却配了青岛的，代码跑起来可能报 NoSuchBucket 或 AccessDenied。endpoint可以在Bucket详情页找到，不要自己拍脑袋写。
3. 核心代码实战：从单文件到分片上传
3.1 最基础的put_object，三行代码跑通上传
最简的上传方式就是用put_object，把本地文件或者内存里的bytes一次性传上去。
import oss2 access_key_id = "你的AccessKey ID" access_key_secret = "你的AccessKey Secret" endpoint = "oss-cn-hangzhou.aliyuncs.com" bucket_name = "your-bucket" auth = oss2.Auth(access_key_id, access_key_secret) bucket = oss2.Bucket(auth, endpoint, bucket_name) # 方式一：上传本地文件 result = bucket.put_object_from_file("images/avatar.jpg", "/tmp/avatar.jpg") print(result.status) # 方式二：上传bytes数据 content = b"hello oss" result = bucket.put_object("test.txt", content) print(result.status)put_object_from_file的第一个参数是OSS上的对象名（object key），可以带目录前缀，比如images/avatar.jpg，OSS会自动在逻辑上创建目录结构。第二个参数是本地文件路径。返回值result.status是200表示成功。
这里有几个隐藏细节。第一，对象名别以/开头，否则访问URL会多一层路径，增加出错概率。第二，上传中文文件名时，OSS会保留原始名称，但URL访问需要做URL编码，所以建议在代码里统一把文件名转成不带中文的规范化名称，例如用uuid+ 扩展名。第三，如果文件不存在，put_object_from_file会抛FileNotExistError，建议先用os.path.exists判断。
3.2 流式上传：不落盘，直接把内存数据传上去
很多场景下，文件不是从本地磁盘来的，而是从网络下载、从数据库读取、或者从另一个接口的响应流中拿到。如果先写到临时文件再上传，多一次磁盘IO，还要清理临时文件，费时费力。这种场景直接用流式上传。
import requests import oss2 auth = oss2.Auth("你的AccessKey ID", "你的AccessKey Secret") bucket = oss2.Bucket(auth, "oss-cn-hangzhou.aliyuncs.com", "your-bucket") # 从远程URL下载图片并直接上传到OSS response = requests.get("https://example.com/some-image.jpg", stream=True) result = bucket.put_object( "images/cache.jpg", response.raw ) print(result.status)同时，也可以把文件对象传入。Flask或Django里接收上传文件时，文件本身就是一个类文件对象，直接传给SDK即可：
from flask import Flask, request app = Flask(__name__) @app.route("/upload", methods=["POST"]) def upload(): f = request.files["file"] result = bucket.put_object("uploads/" + f.filename, f.stream) if result.status == 200: return {"code": 0, "url": f"https://your-bucket.oss-cn-hangzhou.aliyuncs.com/uploads/{f.filename}"} return {"code": 500, "msg": "upload failed"}, 500流式上传的好处是内存可控，不会因为文件太大导致进程内存暴涨。访问上传接口的方式是：OSS会直接用流读取方式把数据写入，整个过程不会为文件内容开辟大块内存。这对于视频录制、日志文件这类动态生成内容的场景尤其合适。
3.3 大文件分片上传：超过100MB怎么办
我见过一个案例：用户直接调put_object上传一个1.2GB的视频，跑到一半进程被杀，OSS上多了一个“碎片”对象，还得手动清理碎片。为什么？因为简单上传接口一次性读入文件内容，大文件要么QPS受限、要么内存吃紧。正确解法是分片上传。
分片上传的核心逻辑：把文件切成多个部分，分别上传，最后调用complete，OSS自动拼接成完整文件。好处是：任意一片失败只需重传该片，不需要从头再来；且可以并发上传多个分片，速度更快。
import oss2 from oss2 import determine_part_size auth = oss2.Auth("你的AccessKey ID", "你的AccessKey Secret") bucket = oss2.Bucket(auth, "oss-cn-hangzhou.aliyuncs.com", "your-bucket") # 文件总大小 total_size = os.path.getsize("/path/to/big-file.zip") # 确定分片大小，默认给100KB到10MB之间的值，具体根据文件大小计算 part_size = determine_part_size(total_size, preferred_size=100 * 1024) # 初始化分片上传，拿到upload_id upload_id = bucket.init_multipart_upload("backup/big-file.zip").upload_id # 分片读取并上传 with open("/path/to/big-file.zip", "rb") as f: part_number = 1 parts = [] while True: chunk = f.read(part_size) if not chunk: break result = bucket.upload_part("backup/big-file.zip", upload_id, part_number, chunk) parts.append(oss2.models.PartInfo(part_number, result.etag)) part_number += 1 # 可以在这里打印进度 uploaded = (part_number - 1) * part_size print(f"progress: {min(uploaded, total_size) / total_size:.2%}") # 合并分片 result = bucket.complete_multipart_upload("backup/big-file.zip", upload_id, parts) print(result.status)注意几点：分片编号要从1开始，不能跳号；保存每个分片的etag，合并时会用到；分片大小有下限要求（OSS最低是100KB），太小会报错；如果中途失败，记得调用bucket.abort_multipart_upload终止，否则会产生碎片计费。
3.4 断点续传：网络不稳也能救回来
分片上传解决了大文件问题，但还没有解决“网络中断后从断点继续”的问题。oss2提供了resumable_upload，封装了分片上传加本地点位记录的机制。
import oss2 auth = oss2.Auth("你的AccessKey ID", "你的AccessKey Secret") bucket = oss2.Bucket(auth, "oss-cn-hangzhou.aliyuncs.com", "your-bucket") result = bucket.resumable_upload( "backup/big-file.zip", "/path/to/big-file.zip", store=oss2.ResumableStore(root="/tmp/oss_store"), part_size=100 * 1024, num_threads=4 ) print(result.status)这个接口在做的事情：把上传进度和已上传分片信息记录在本地目录store.root下，一旦调用被中断，下次调用同一个object key和本地文件时，会自动跳过已完成的分片，只传剩余部分。观看大量实现细节后，你会发现它和无缝断点续传的核心区别在于，记录文件本身需要及时落盘，才能保证下次启动时知道哪些分片已完成。
实际使用中，num_threads参数可以用来控制并发度。并发太高会触发OSS的QPS限制，太低又慢，建议4-8。此外，store目录在容器环境下要用持久化卷，否则Pod重启后点位就丢了。
4. 工程化封装：写一个能上生产的OSS上传工具类
4.1 统一封装，上下游调用方不用各写各的
项目里直接到处调用SDK会导致一个糟糕的局面：每个人对endpoint、超时、重试策略的理解都不同，出了问题排查靠运气。我习惯把OSS上传封装成一个工具类，对外只暴露几个方法。
import os import uuid import oss2 from oss2 import ResumableStore, determine_part_size class OSSClient: def __init__(self, access_key_id, access_key_secret, endpoint, bucket_name): self.auth = oss2.Auth(access_key_id, access_key_secret) self.bucket = oss2.Bucket(self.auth, endpoint, bucket_name) def _gen_object_name(self, prefix, ext): return f"{prefix}/{uuid.uuid4().hex}{ext}" def upload_bytes(self, data, prefix="files", ext=""): object_name = self._gen_object_name(prefix, ext) result = self.bucket.put_object(object_name, data) if result.status == 200: return object_name raise RuntimeError(f"upload fail, status={result.status}") def upload_file(self, local_path, prefix="files"): ext = os.path.splitext(local_path)[1] object_name = self._gen_object_name(prefix, ext) # 超过200MB走分片，否则直接传 file_size = os.path.getsize(local_path) if file_size > 200 * 1024 * 1024: result = self.bucket.resumable_upload( object_name, local_path, store=ResumableStore(root="/tmp/oss_store"), num_threads=4 ) else: result = self.bucket.put_object_from_file(object_name, local_path) if result.status == 200: return object_name raise RuntimeError(f"upload fail, status={result.status}")问题来了：uuid会不会导致文件名太长，失去可读性？如果你有业务上需要检索文件来源的需求，可以在文件名里加语义前缀，比如order/20250601/2f3a...png。实际上我更推荐对象名中加入日期分层，即prefix/yyyy/mm/dd/uuid.ext，这样管理控制台和日志里查文件都方便，也天然避免了单个目录下对象数量过多的问题。
4.2 回调机制：让OSS主动通知你的业务系统
很多时候，文件上传完成后，业务系统需要立刻知道并把记录写入数据库。传统的做法是上传成功后，业务代码自己同步落库。但如果是前端直传OSS，后端完全不知道文件什么时候传完，这时候就需要OSS回调（Callback）。
实现思路：前端直传时附带一个回调参数，OSS上传完成后会向后端指定的接口发一个HTTP POST请求，携带自定义参数和上传信息。Python后端接收回调的Flask接口示例：
import base64 import json import requests from flask import Flask, request app = Flask(__name__) @app.route("/oss/callback", methods=["POST"]) def oss_callback(): # 1. 验证签名（生产环境必须做，这里简化） # 2. 读取回调参数 body = request.get_data() params = request.form # params.get("object") 即object key # params.get("bucket") 即bucket名称 # 业务自定义字段会原样带回 # 这里可以把文件信息写入数据库、触发异步任务等 print("callback received:", params) # 3. 返回OSS规定的JSON格式 resp_body = json.dumps({"Status": "OK"}) resp_base64 = base64.b64encode(resp_body.encode()).decode() return f'{{"Status":"OK"}}', 200, {"Content-Type": "application/json"}回调接口必须返回特定格式，OSS才会认为回调成功；如果签名校验失败，OSS会认为上传无效。而且要注意，回调地址不能是内网地址，OSS服务端无法访问你的内网。这个机制用好了，可以在前端直传场景下，实现“用户传完文件，后端立刻更新数据库”的无缝衔接。
4.3 带进度条的上传：别让用户干等
用户上传大文件时最怕“无响应”，前端需要展示进度条，而后端如果处理上传，也需要拿到实时进度。oss2提供了进度回调函数：
import oss2 auth = oss2.Auth("你的AccessKey ID", "你的AccessKey Secret") bucket = oss2.Bucket(auth, "oss-cn-hangzhou.aliyuncs.com", "your-bucket") def progress_callback(consumed_bytes, total_bytes): if total_bytes: rate = consumed_bytes / total_bytes print(f"progress: {rate * 100:.2f}%") bucket.put_object_from_file( "files/large.zip", "/path/to/large.zip", progress_callback=progress_callback )在我的实践里，服务端场景我通常只写日志，不把进度实时推到前端，因为服务端进程和前端请求之间还要走一层消息推送，增加复杂度。但内网工具、批处理脚本这种场景，打印进度对排查很有帮助，能直观看出卡在哪一段。如果做前端直传，进度条完全由前端基于XMLHttpRequest的upload.onprogress回调实现，不占用后端任何资源，这也是我推荐直传方案的原因之一。
5. 线上问题排查实录
5.1 AccessDenied：权限、Bucket和Region逐个查
线上最常见的错误就是 AccessDenied，服务端返回403。绝大多数情况不是AccessKey错了，而是权限没配好。
按顺序排查：第一，确认这个AccessKey对应的账号有没有操作目标Bucket的权限，RAM用户的权限策略里Resource字段是否正确。第二，确认Bucket是私有的还是公共读，私有Bucket直接用URL访问当然403。第三，也是最容易被忽略的，确认endpoint地域和Bucket所在地域一致，跨地域访问会被拒绝。第四，如果你用的是STS临时凭证，检查Expiration字段，临时凭证过期了同样会报AccessDenied。
还有一个隐藏点：Bucket有“跨域设置”（CORS）时，如果Ancestor字头错误，浏览器中直传会报错，但Postman/脚本不见得暴露问题。所以遇到浏览器端403、服务端正常的情况，优先检查Bucket的CORS规则。
5.2 SignatureDoesNotMatch：时间戳和签名错位
签名错误，这类报错看起来神秘，排查方向却很清晰。OSS签名机制会把请求时间、object名、访问密钥等拼串后做HMAC-SHA1计算，任何一项不一致就会签名不匹配。
最常见的原因：服务器本地时间或客户端系统时间与真实时间偏差太大。OSS要求请求时间和服务端时间差在15分钟以内，超了这个窗口直接拒绝。排查时先date看服务器时间，如果偏了就同步。另一类原因是代码里手动拼接了签名串，拼错了字段。用官方SDK时基本不会遇到，因为SDK封装好了签名逻辑，所以不建议自己造轮子去攻击签名。自建签名多出问题，最后还得回来用官方库。
5.3 大文件上传慢/超时的排查
上传大文件时出现RequestTimeout或进度长时间不动，通常不是OSS问题，而是链路问题。
先判断网络环境。云服务器上传OSS走内网还是公网，走公网会有带宽上限，特别是按固定带宽计费的实例，上传流量也可能受限制。如果业务服务器和OSS在同一个地域，域名可以换成内网地址（例如oss-cn-hangzhou-internal.aliyuncs.com），速度快、免流量费。这个优化我每次都会做，效果立竿见影。
再判断是否启用了代理。有些公司网络环境设置了HTTP代理，大量数据传输会被代理拦截或卡住，导致上传超时。排查时看一下环境变量里是否有http_proxy、https_proxy，如果SDK默认走了代理，几十GB文件直接断。
还有一点容易被忽略：DNS解析问题。解析到一层CDN或防火墙IP时，请求迟迟不出去。可以用dig或nslookup看下解析结果，必要时在SDK初始化时指定CName参数，走自定义域名。
5.4 计费和限流的那些坑
OSS不是免费的，很多人上传完看账单才发现费用超预期。费用主要由三块构成：存储容量费、请求次数费、流量费。
存储容量费按天计费，私有Bucket和低频访问的单价不同，但注意，你在控制台删除了文件，如果还有分片上传产生的碎片（Multipart Upload引发的碎片对象），这些碎片照样计费。我遇到过同事用分片上传跑批任务，每次失败都在Bucket里留一堆碎片，一个月后存储量突增。解决办法是写一个清理脚本，定期调用list_multipart_uploads拿到upload_id列表，然后逐个abort_multipart_upload。
import oss2 auth = oss2.Auth("你的AccessKey ID", "你的AccessKey Secret") bucket = oss2.Bucket(auth, "oss-cn-hangzhou.aliyuncs.com", "your-bucket") # 列出当前bucket下所有未完成的分片上传 for upload in oss2.MultiPartUploadIterator(bucket): object_name = upload.key upload_id = upload.upload_id bucket.abort_multipart_upload(object_name, upload_id) print(f"aborted: {object_name}, upload_id={upload_id}")请求次数费很多人忽略。OSS默认按请求次数计费，每万次请求几毛钱，看起来不多，但海量小文件上传时，一万次就是一万个Object，如果脚本里循环调用单文件上传，日积月累也是一笔钱。相比于后来多一个大IO和前置排查的时间，优化手段就是把小文件合并打包上传，或者用分段并发把QPS提上去。当然，QPS也不能无限提，OSS对单Bucket的QPS有限制，超过后会返回SlowDown，遇到这个错误需要退避重试，SDK里默认有重试逻辑，生产环境建议自定义退避策略：指数退避 + 最大重试次数。
比如，在初始化Bucket时传入oss2.defaults.connection_pool_size = 20提高连接池大小，可以避免高并发上传时建立连接耗时过高。同时在代码里对oss2.exceptions.ServerError做捕获和重试：
import time import oss2 from tenacity import retry, stop_after_attempt, wait_exponential auth = oss2.Auth("你的AccessKey ID", "你的AccessKey Secret") bucket = oss2.Bucket(auth, "oss-cn-hangzhou.aliyuncs.com", "your-bucket") @retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, max=10)) def safe_upload(object_name, file_path): result = bucket.put_object_from_file(object_name, file_path) if result.status != 200: raise RuntimeError(f"upload failed: {result.status}") return result safe_upload("files/data.txt", "/tmp/data.txt")这段代码里用tenacity库做重试，遇到异常会自动等待且指数退避，5次重试足够应对绝大多数瞬时性错误。别在每次上传失败后直接抛异常，那会让上游调用方频繁重试整个业务。
结尾再分享一个实际的体会：我最早接触OSS上传时，总觉得SDK就是几十行代码的事，真到了生产环境才发现大头是权限设计、分片策略、断点续传、回调通知、计费可见性这些“看不见的细节”。上面这些坑，几乎都是我在线上实战里一个个踩出来的。你把这套代码和思路吸收进去，再去设计自己项目的上传模块，起码能少走一半弯路。最后一个小技巧：上线前务必把Buckeet权限设置为“私有读+签名URL访问”，不要图方便设成公共读，不然一张图被刷流量，月底账单能让你肉疼。
