小编Gre*_*0ry的帖子

我一直在玩SQLAlchemy,发现我无法可靠地跟踪数据库中的更改内容.

我创建了一个例子来解释我的担忧:

import re
import datetime

from sqlalchemy import create_engine

from sqlalchemy.ext.declarative import (
    declarative_base,
    declared_attr,
    )

from sqlalchemy import (
    create_engine,
    event,
    Column,
    Boolean,
    Integer,
    String,
    Unicode,
    DateTime,
    Index,
    ForeignKey,
    CheckConstraint,
    )

from sqlalchemy.orm import (
    scoped_session,
    sessionmaker,
    Session,
    relationship,
    backref,
    )

import transaction

from zope.sqlalchemy import ZopeTransactionExtension

class ExtendedSession(Session):
    my_var = None

DBSession = scoped_session(
    sessionmaker(extension=ZopeTransactionExtension(),
        class_=ExtendedSession
        )
    )

class BaseModel(object):
    query = DBSession.query_property()

    id = Column(
        Integer,
        primary_key=True,
        )

    @declared_attr
    def __tablename__(cls):
        class_name = re.sub(r"([A-Z])", r"_\1", cls.__name__).lower()[1:] …

我创建了一个 Kinesis 数据流，并将格式如下的记录输入其中：

{
    "ufo": 5,
    "unicorn": 7,
    "something else": 9,
}

我正在尝试编写一个转换 lambda，它将转换上述记录并将三个记录插入到 Redshift 中。Redshift表架构是这样的：

CREATE TABLE my_table(
    timestamp DATETIME,
    name TEXT,
    count INT
) SORTKEY(timestamp);

因此本质上，我需要从单个 Kinesis 记录在 Redshift 中生成三行。

转换 lambda 的文档没有解释我想要实现的目标是否可能。我已经尝试在编码之前将所有有效负载包装在数组中，但这会导致 Redshift 端出现处理错误。我还尝试返回具有相同记录 ID 的多个记录 - 这也会导致错误。