如何获取 SQLAlchemy 自定义类型的原始值？

Question

简而言之我的问题：是否可以在 SQLAlchemy 中访问自定义类型的原始值？

我对自定义类型的实现：

class JSONDocument(TypeDecorator):
    # override field
    impl = JSON

    # Write
    def process_bind_param(self, value, dialect):
        if value is not None:
            req = requests.post(url, data=json.dumps({'data': value}))
            uri = req.json()['uri']
            value = uri
        return value

    # Read
    def process_result_value(self, value, dialect):
        if value is not None:
            uri = value
            req = requests.get(uri)
            value = req.json()['data']
        return value

我的桌子模型：

class SomeTable(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    # The column with large json data
    json_data = db.Column(JSONDocument)

每当我尝试更新或插入一行，结果将写入json_data 列时，SQLAlchemy 将调用JSONDocument 的process_bind_param。然后它会执行一个 POST API 调用，在我的 NoSQL 表中创建一个新的数据条目，并将我的 SQL 表中的 json_data 列更新为返回的 URI。

每当我尝试访问 json_data 列时，它都会对存储在我的 SQL 表中的 URI 执行 GET API 调用，并返回存储在我的 NoSQL 表中的实际 JSON 数据。

我的问题：如何获取存储在JSONDocument 覆盖的SQL 表中的URI？每当我更新 json_data 时，我都想对旧 URI 执行 DELETE API 调用。

我已经尝试使用 SQLAlchemy 的 Mapping 事件回调，但我仍然无法获取自定义类型背后的 URI。

如果我的问题问得不够清楚，请告诉我。任何建议或想法表示赞赏。

我正在尝试做的任务的总结：

我有一个包含以下列的 Postgresql 数据库 (RDS) 表：id(int)、json_data(string)、other_columns...

由于json_data 列占用了太多存储空间，我正在努力将该列中的数据移动到其他 NoSQL 数据库 (DynamoDB)。我实现了一个 RESTful API 服务器，用于在我的 NoSQL 表中更新、插入和软删除数据。

我想实现一个 SQLAlchemy 自定义类型，当我的应用尝试访问其他 Python 代码中的 json_data 列时，它会自动调用我的 API 方法。

Answer 1

您是否考虑过摆脱自定义 TypeDecorator 并将您的逻辑移动到类似于使用属性的 SomeTable？

class SomeTable(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    # The column with large json data
    _uri = db.Column(db.JSON)

    @property
    def json_data(self):
        if self._uri is not None:
            req = requests.get(self._uri)
            return req.json()['data']
        else:
            return None

    @json_data.setter
    def json_data(self, value):
        req = requests.post(url, data=json.dumps({'data': value}))
        self._uri = req.json()['uri']

    @property
    def uri(self):
        return self._uri