0

我需要创建序列，但在一般情况下不使用Sequence类。

USN = Column(Integer, nullable = False, default=nextusn, server_onupdate=nextusn)

，此功能nextusn需要func.max(table.USN)在模型中生成行的值。

我尝试使用这个

class nextusn(expression.FunctionElement):
    type = Numeric()
    name = 'nextusn'

@compiles(nextusn)
def default_nextusn(element, compiler, **kw):
    return select(func.max(element.table.c.USN)).first()[0] + 1

但在此上下文中的元素不知道element.table。存在解决此问题的方法吗？

4

1 回答 1

0

这有点棘手，原因如下：

  1. 如果表为空，您的 SELECT MAX() 将返回 NULL；您应该使用 COALESCE 来生成默认的“种子”值。见下文。

  2. 使用 SELECT MAX 插入行的整个方法对于并发使用完全不安全-因此您需要确保一次只在表上调用一个 INSERT 语句，否则您可能会违反约束（您绝对应该有一些约束本栏目中的种类）。

  3. 从 SQLAlchemy 的角度来看，您需要自定义元素来了解实际的 Column 元素。我们可以通过在事后将“nextusn()”函数分配给 Column 来实现这一点，或者在下面我将展示一个使用事件的更复杂的方法。

  4. 我不明白您要使用“server_onupdate=nextusn”做什么。SQLAlchemy 中的“server_onupdate”实际上并没有为您运行任何 SQL，例如，如果您创建了一个触发器，这是一个占位符；但“SELECT MAX(id) FROM table”也是一个 INSERT 模式，我不确定你的意思是在 UPDATE 上发生任何事情。

  5. @compiles 扩展需要返回一个字符串，通过compiler.process() 在那里运行select()。见下文。

例子：

from sqlalchemy import Column, Integer, create_engine, select, func, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql.expression import ColumnElement
from sqlalchemy.schema import ColumnDefault
from sqlalchemy.ext.compiler import compiles
from sqlalchemy import event

class nextusn_default(ColumnDefault):
    "Container for a nextusn() element."
    def __init__(self):
        super(nextusn_default, self).__init__(None)

@event.listens_for(nextusn_default, "after_parent_attach")
def set_nextusn_parent(default_element, parent_column):
    """Listen for when nextusn_default() is associated with a Column,
    assign a nextusn().

    """
    assert isinstance(parent_column, Column)
    default_element.arg = nextusn(parent_column)


class nextusn(ColumnElement):
    """Represent "SELECT MAX(col) + 1 FROM TABLE".
    """
    def __init__(self, column):
        self.column = column

@compiles(nextusn)
def compile_nextusn(element, compiler, **kw):
    return compiler.process(
                select([
                    func.coalesce(func.max(element.column), 0) + 1
                ]).as_scalar()
            )

Base = declarative_base()

class A(Base):
    __tablename__ = 'a'

    id = Column(Integer, default=nextusn_default(), primary_key=True)
    data = Column(String)

e = create_engine("sqlite://", echo=True)

Base.metadata.create_all(e)

# will normally pre-execute the default so that we know the PK value
# result.inserted_primary_key will be available
e.execute(A.__table__.insert(), data='single row')

# will run the default expression inline within the INSERT
e.execute(A.__table__.insert(), [{"data": "multirow1"}, {"data": "multirow2"}])

# will also run the default expression inline within the INSERT,
# result.inserted_primary_key will not be available
e.execute(A.__table__.insert(inline=True), data='single inline row')
于 2013-05-03T15:53:17.087 回答