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)#orinstance = db.session.query(EnrollmentFormData).get(1)print(instance.create_presigned_url)#orprint([x.create_presigned_url for x in db.session.query(EnrollmentFormData).all()]