SqlAlchemy 未定义此多态标识符

2 投票
1 回答
5031 浏览
提问于 2025-04-18 09:01

我正在尝试存储用户在第三方网站上的oauth信息,BaseConnection表用来存储连接的基本信息。

class BaseConnection(Base):
    __tablename__ = "BaseConnection"
    id = Column(Integer, primary_key=True)
    username = Column(Unicode(255),
                      ForeignKey('users.username'))
    type = Column(Unicode(255))
    is_active = Column(Boolean, default=False)
    last_connection = Column(DateTime, onupdate=datetime.utcnow)

    __mapper_args__ = {
        'polymorphic_identity': 'BaseConnection',
        'polymorphic_on': type
    }

    def __init__(self, username, type, is_active):
        self.username = username
        self.type = type
        self.is_active = is_active

Twitter连接用来存储Twitter账户的oauth信息。

class TwitterConnection(Base):
    __tablename__ = "TwitterConnection"

    connection_id = Column(Integer, ForeignKey('BaseConnection.id'),
                           primary_key=True)
    uid = Column(Unicode(255))
    access_key = Column(Unicode(255))
    access_secret = Column(Unicode(255))
    twitter_username = Column(Unicode(255))
    refresh_date = Column(DateTime)

    __mapper_args__ = {
        'polymorphic_identity': 'TwitterConnection'
    }

    def __init__(self, connection_id, 
                 uid, access_key, access_secret, twitter_username):
        self.connection_id = connection_id
        self.uid = uid
        self.access_key = access_key
        self.access_secret = access_secret
        self.twitter_username = twitter_username
        self.refresh_date = datetime.now()

然后我使用代码将第一个信息添加到表中。

base_connection = BaseConnection('admin', TwitterConnection.__name__, True)
base_connection.last_connection = datetime.now()
DBSession.add(base_connection)
DBSession.flush()
tconnection = TwitterConnection(
                 base_connection.id,
                 uid=u'1234',
                 access_key=u'dummy',
                 access_secret=u'dummy',
                 twitter_username='dummy')
DBSession.add(tconnection)
DBSession.flush()
transaction.commit()

我正在尝试写一个查询,找出特定用户名的用户所有第三方账户。

qry = BaseConnection.query.filter(
                BaseConnection.username==username)
qry.all()

上面的查询出现了错误。

`AssertionError: No such polymorphic_identity u'TwitterConnection' is defined

我对sqlalchemy还很陌生,任何帮助都非常感谢,谢谢你。

1 个回答

3

抱歉,

class TwitterConnection(Base):

应该是

class TwitterConnection(BaseConnection):

撰写回答