Python Django应用复用AWS角色认证及视图报错解决求助

问题背景 #

我正在开发一个基于Python 3.6和Django 1.10的项目，需要使用终端用户的AWS账户资源将代码部署到其账户中。已经配置了两个账户（应用账户与用户账户）并完成认证，但在另一个视图复用该认证时出现了错误。

错误信息：

AwsAuthentication is missing a QuerySet. Define AwsAuthentication.model, AwsAuthentication.queryset, or override AwsAuthentication.get_queryset().

我的尝试代码（views.py）：

def boto3_with_role(role_arn, session_prefix, external_id, **kwargs):
    """
    Create a partially applied session to assume a role with an external id.
    A unique session_name will be generated by {session_prefix}_{time}
    `session` can be passed, otherwise the default sesion will be used
    """
    sts = boto3.client('sts')
    res = sts.assume_role(
        RoleArn=role_arn,
        RoleSessionName='{}_{}'.format(session_prefix, int(time.time())),
        ExternalId=external_id,
        *kwargs,
    )
    creds = res['Credentials']
    print(creds)
    return partial(boto3.session.Session,
                   aws_access_key_id=creds['AccessKeyId'],
                   aws_secret_access_key=creds['SecretAccessKey'],
                   aws_session_token=creds['SessionToken'],
                   )

class AwsAuthentication(LoginRequiredMixin, CreateView):
    def post(self, request, *args, **kwargs):
        AwsSession = boto3_with_role('arn:aws:iam::497736713165:role/IstiocloudCrossAccount', 'MyPrefix', 'abd37214@cloud')
        my_session = AwsSession()
        client = my_session.resource('s3')
        for bucket in client.buckets.all():
            print(bucket.name)
        return render(request, 'dockerDep/aws/selectDeployment.html', {'bucket': bucket.name})

# 从此处开始AWS部署
class AwSlsDeployment(LoginRequiredMixin, CreateView):
    def get(self, request, *args, **kwargs):
        AwsSession = boto3_with_role('arn:aws:iam::497736713165:role/IstiocloudCrossAccount', 'MyPrefix', 'abd37214@cloud')
        user_session = AwsSession()
        client = user_session.resource('s3')
        for bucket in client.buckets.all():
            print(bucket.name)
        return render(request, 'dockerDep/aws/slsDeployment.html', {'bucket': bucket.name})

我需要解决两个问题：一是修复这个QuerySet缺失的错误，二是如何在多个视图中复用boto3_with_role()的认证逻辑？

解决方案 #

1. 先修复QuerySet缺失的错误

这个错误的根源很清晰：你使用了CreateView但没有指定对应的模型或QuerySet。CreateView是Django用来创建数据库模型实例的通用视图，它默认要求明确要操作的模型，所以会强制你定义model、queryset或重写get_queryset()方法。

但你的AwsAuthentication和AwSlsDeployment视图根本不需要创建模型实例——它们只是处理AWS认证并获取S3桶数据。所以正确的做法是把这两个视图的父类从CreateView改成普通的View：

修改后的AwsAuthentication视图：

class AwsAuthentication(LoginRequiredMixin, View):  # 替换CreateView为View
    def post(self, request, *args, **kwargs):
        AwsSession = boto3_with_role('arn:aws:iam::497736713165:role/IstiocloudCrossAccount', 'MyPrefix', 'abd37214@cloud')
        my_session = AwsSession()
        client = my_session.resource('s3')
        # 修复：循环后只保留最后一个桶的问题，改为收集所有桶名列表
        bucket_names = [bucket.name for bucket in client.buckets.all()]
        print(bucket_names)
        return render(request, 'dockerDep/aws/selectDeployment.html', {'buckets': bucket_names})

同样修改AwSlsDeployment视图：

class AwSlsDeployment(LoginRequiredMixin, View):  # 替换CreateView为View
    def get(self, request, *args, **kwargs):
        AwsSession = boto3_with_role('arn:aws:iam::497736713165:role/IstiocloudCrossAccount', 'MyPrefix', 'abd37214@cloud')
        user_session = AwsSession()
        client = user_session.resource('s3')
        bucket_names = [bucket.name for bucket in client.buckets.all()]
        print(bucket_names)
        return render(request, 'dockerDep/aws/slsDeployment.html', {'buckets': bucket_names})

2. 优雅复用Boto3认证逻辑

现在要解决重复代码的问题——两个视图里都重复调用了boto3_with_role()并创建session。我们可以用Django Mixin来封装这个逻辑，让需要AWS访问的视图直接继承这个Mixin即可。

第一步：创建AWS认证Mixin

在你的views.py（或者单独的mixins.py文件）中定义一个Mixin类：

from django.conf import settings
import boto3
from functools import partial
import time

class AwsSessionMixin:
    # 从settings中读取配置，更易于维护
    AWS_ROLE_ARN = settings.AWS_CROSS_ACCOUNT_ROLE_ARN
    AWS_SESSION_PREFIX = settings.AWS_SESSION_PREFIX
    AWS_EXTERNAL_ID = settings.AWS_EXTERNAL_ID

    def get_aws_session(self):
        """获取已认证的AWS Session，复用boto3_with_role逻辑"""
        AwsSession = boto3_with_role(
            self.AWS_ROLE_ARN,
            self.AWS_SESSION_PREFIX,
            self.AWS_EXTERNAL_ID
        )
        return AwsSession()

同时在settings.py中添加配置：

# settings.py
AWS_CROSS_ACCOUNT_ROLE_ARN = 'arn:aws:iam::497736713165:role/IstiocloudCrossAccount'
AWS_SESSION_PREFIX = 'MyPrefix'
AWS_EXTERNAL_ID = 'abd37214@cloud'

第二步：在视图中复用Mixin

修改你的两个视图，让它们继承这个Mixin，然后直接调用get_aws_session()：

class AwsAuthentication(LoginRequiredMixin, AwsSessionMixin, View):
    def post(self, request, *args, **kwargs):
        my_session = self.get_aws_session()  # 直接复用Mixin的认证逻辑
        client = my_session.resource('s3')
        bucket_names = [bucket.name for bucket in client.buckets.all()]
        return render(request, 'dockerDep/aws/selectDeployment.html', {'buckets': bucket_names})

class AwSlsDeployment(LoginRequiredMixin, AwsSessionMixin, View):
    def get(self, request, *args, **kwargs):
        user_session = self.get_aws_session()  # 直接复用Mixin的认证逻辑
        client = user_session.resource('s3')
        bucket_names = [bucket.name for bucket in client.buckets.all()]
        return render(request, 'dockerDep/aws/slsDeployment.html', {'buckets': bucket_names})

这样一来，所有需要AWS认证的视图都可以通过继承AwsSessionMixin来复用认证逻辑，不需要重复写boto3_with_role()的调用代码。如果以后需要修改AWS配置参数，只需要修改settings.py即可，非常方便。

额外优化建议

• 添加错误处理逻辑：捕获boto3的认证错误（比如权限不足、角色不存在等），返回友好的提示给用户，避免直接抛出异常。
• 缓存AWS Session：如果短时间内多次调用，可以考虑缓存生成的Session，减少STS角色切换的请求次数。

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