唯一序列号到列

Question

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

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

，这funcion 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。存在的方式来解决这个问题？

Answer 1

这是一个有点棘手，原因如下：

1. 你的SELECT MAX（）将返回NULL，如果表是空的;你应该使用COALESCE来产生默认的“种子”值。见下文。

2. 用SELECT MAX插入行的整个方法对于并发使用是完全不安全的 - 因此您需要确保一次只能在表上调用一个INSERT语句，否则您可能会受到约束违规（您应该明确在这一列上有某种限制）。

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

4. 我不明白“server_onupdate = nextusn”会带来什么。 SQLAlchemy中的“server_onupdate”实际上并没有为你运行任何SQL，如果你创建了一个触发器，这是一个占位符;而且“SELECT MAX（id）FROM table”是一个INSERT模式，我不确定你的意思是在UPDATE发生什么。

5. @compiles扩展需要返回一个字符串，通过compile.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')