Sqlalchemy - 创建 Amazon S3 预签名 URL

Question

我正在尝试创建一个 S3 预签名 URL，以便人们可以临时访问存储在 S3 存储桶中的文件。我有一个这样的函数来创建 url：def create_presigned_url(file_name):        """Generate a presigned URL to share an S3 object            return: Presigned URL as string. If error, returns None.        """                bucket_name = 'mybucket'        object_name = 'pdfs/{}'.format(file_name)        expiration = 3600        # Generate a presigned URL for the S3 object        try:            response = bucket_resource.generate_presigned_url(                "get_object",                Params={"Bucket": bucket_name, "Key": object_name},                ExpiresIn=expiration,            )        except Exception as e:            return None            # The response contains the presigned URL        return response现在我想将此 url 应用于下面定义的表 EnrollmentFormData：class EnrollmentFormData(db.Model):    __tablename__ = 'enrollment_form_data'    id = db.Column(db.Integer, primary_key=True)    full_name = db.Column(db.String(300), nullable=False)    employee_number = db.Column(db.String(12),nullable=False)    department_code = db.Column(db.String(12), nullable=False)    sign_image = db.Column(db.BLOB, nullable=False)    aws_path = db.Column(db.String(2083), nullable=False)    file_name = db.Column(db.String(2083), nullable=False)    timestamp = db.Column(db.DateTime, nullable=False, default=lambda:     datetime.datetime.now(pytz.timezone("America/New_York")))我尝试了两件事：首先，我尝试将列名作为函数的参数传递： query = db.session.query(        EnrollmentFormData.id,        create_presigned_url(EnrollmentFormData.file_name),        EnrollmentFormData.timestamp    ).all()这可能不起作用，因为该函数无法使用 SqlAlchemy 类型。以上也不行两者都给出的错误是：SQL expression, column, or mapped entity expected - got ''抱歉无法粘贴全部内容，因为我不想泄露访问密钥。任何人都可以帮助完成它吗？

Answer 1

除非您打算提供自定义 SQL 函数，否则您将希望避免使用.expression 修饰符并继续使用实例级方法。

    @hybrid_property

    def create_presigned_url(self):

        """Generate a presigned URL to share an S3 object

            return: Presigned URL as string. If error, returns None.

        """

        

        bucket_name = 'myBucket'

        object_name = 'pdfs/{}'.format(self.file_name)

        expiration = 3600

        # Generate a presigned URL for the S3 object

        try:

            response = bucket_resource.generate_presigned_url(

                "get_object",

                Params={"Bucket": bucket_name, "Key": object_name},

                ExpiresIn=expiration,

            )

        except Exception as e:

            return None

    

        # The response contains the presigned URL

        return response

要使用它，您需要在实例级别而不是类级别进行操作。所以你的查询是针对实例的：

instance = db.session.query(EnrollmentFormData).filter(EnrollmentFormData.id==1).first()

print(instance.create_presigned_url)

#or

instance = db.session.query(EnrollmentFormData).get(1)

print(instance.create_presigned_url)

#or

print([x.create_presigned_url for x in db.session.query(EnrollmentFormData).all()]