Django中通过annotate获取私有S3媒体文件预签名URL方案

首先明确一个核心前提：S3预签名URL依赖AWS SigV4签名算法、你的账号访问密钥以及自定义过期时间，无法通过纯数据库层函数直接生成——PostgreSQL没有内置AWS签名逻辑，也绝对不建议把AWS密钥存入数据库做签名计算，会带来极高的安全风险。
你之前公开桶的方案是纯字符串拼接，所以可以直接在数据库层用Concat完成；私有桶的签名逻辑必须在应用层执行，我们可以通过自定义QuerySet逻辑实现和原有annotate几乎一致的调用体验，完全不需要引入序列化器，也不影响你后续用values()取数。

可复用实现代码 #

以下实现基于你现有的技术栈（PostgreSQL + boto3 + django-storages），对外调用方式和你之前的getImageFormatter几乎无差别。
首先初始化全局S3存储实例，避免重复创建连接：

from django.db.models import Case, When, Value, CharField, F
from django.conf import settings
from storages.backends.s3boto3 import S3Boto3Storage
import functools

# 全局S3存储实例，复用连接
s3_storage = S3Boto3Storage()
# 预签名URL默认过期时间，单位：秒，按需调整
DEFAULT_PRESIGNED_EXPIRY = 3600

方案1：兼容原有函数调用方式 #

直接替换你之前的getImageFormatter函数，不需要修改原有链式调用的逻辑：

def _generate_signed_url(object_key: str, expiry: int = DEFAULT_PRESIGNED_EXPIRY) -> str | None:
    """内部工具：根据S3对象key生成预签名URL，处理空值场景"""
    if not object_key or object_key.strip() == "":
        return None
    try:
        # django-storages 自带的url方法，私有桶配置下会自动生成预签名链接
        return s3_storage.url(name=object_key, expire=expiry)
    except Exception:
        return None

def getImageFormatter(key: str, expiry: int = DEFAULT_PRESIGNED_EXPIRY):
    """
    替换原有公开桶的图片URL格式化函数
    支持后续直接调用values()取数，无序列化器依赖
    """
    annotate_name = f"{key}_url"
    def wrap_queryset(queryset):
        # 第一步：数据库层完成空值过滤，和原有Case逻辑完全一致
        qs = queryset.annotate(
            **{
                annotate_name: Case(
                    When(**{f"{key}__exact": ""}, then=Value(None)),
                    When(**{f"{key}__isnull": False}, then=F(key)),
                    default=Value(None),
                    output_field=CharField(null=True)
                )
            }
        )
        # 第二步：补丁QuerySet迭代逻辑，应用层批量替换成预签名URL
        original_iter = qs._iter
        @functools.wraps(original_iter)
        def patched_iter():
            for row in original_iter():
                # 同时兼容values()返回字典、模型实例两种返回格式
                if isinstance(row, dict):
                    obj_key = row.get(annotate_name)
                    row[annotate_name] = _generate_signed_url(obj_key, expiry)
                else:
                    obj_key = getattr(row, annotate_name, None)
                    setattr(row, annotate_name, _generate_signed_url(obj_key, expiry))
                yield row
        qs._iter = patched_iter
        return qs
    return wrap_queryset

调用方式和你之前的逻辑差异极小：

# 旧公开桶写法
# qs = MyModel.objects.annotate(image_url=getImageFormatter("image")).values("id", "image_url")

# 新私有桶写法
qs = MyModel.objects.all()
qs = getImageFormatter("image")(qs)
result = qs.values("id", "image_url")

方案2：链式调用风格（更推荐） #

如果你希望完全保持ORM链式调用的写法，可以给模型加一个通用的QuerySet方法：

class S3ResourceQuerySet(models.QuerySet):
    def annotate_s3_presigned_url(
        self,
        field_key: str,
        annotate_name: str = None,
        expiry: int = DEFAULT_PRESIGNED_EXPIRY
    ):
        if not annotate_name:
            annotate_name = f"{field_key}_url"
        
        # 数据库层空值预处理
        qs = self.annotate(
            **{
                annotate_name: Case(
                    When(**{f"{field_key}__exact": ""}, then=Value(None)),
                    When(**{f"{field_key}__isnull": False}, then=F(field_key)),
                    default=Value(None),
                    output_field=CharField(null=True)
                )
            }
        )

        # 补丁迭代逻辑，替换预签名URL
        original_iter = qs._iter
        @functools.wraps(original_iter)
        def patched_iter():
            for row in original_iter():
                if isinstance(row, dict):
                    obj_key = row.get(annotate_name)
                    row[annotate_name] = _generate_signed_url(obj_key, expiry)
                else:
                    obj_key = getattr(row, annotate_name, None)
                    setattr(row, annotate_name, _generate_signed_url(obj_key, expiry))
                yield row
        qs._iter = patched_iter
        return qs

给你的模型挂载这个QuerySet之后，就可以直接链式调用，完全符合你要的annotate+values流程：

class MyModel(models.Model):
    image = models.CharField(max_length=500, null=True, blank=True)
    # 其他字段...
    objects = S3ResourceQuerySet.as_manager()

# 直接链式调用，无额外依赖
data = MyModel.objects.filter(xxx=xxx).annotate_s3_presigned_url("image").values("id", "image_url")

注意事项 #

• 性能：单条预签名URL生成耗时约1-2ms，常规列表页（单页100条以内）无感知延迟；如果是超大量数据批量导出场景，可以加一层缓存减少重复签名计算
• 不要尝试用PostgreSQL的PL/Python等扩展在数据库侧实现签名逻辑：一来需要把AWS密钥存入数据库，安全风险极高；二来扩展维护成本高，数据库版本升级很容易出现兼容问题，完全没有必要
• 预签名URL过期时间建议设置在15分钟-1小时区间，不要设置过长，避免URL泄露导致资源被盗刷
• 如果你的django-storages配置正确，私有桶模式下s3_storage.url()会自动生成预签名URL，不需要手动调用boto3的generate_presigned_url方法

内容的提问来源于stack exchange，提问作者unownone