如何使用 sqlalchemist 创建自动生成的雪花值？答案

【问题标题】：how to create a auto-generated value to snowflake using sqlalchemist?如何使用 sqlalchemist 创建自动生成的雪花值？
【发布时间】：2023-02-02 04:20:14
【问题描述】：

我正在尝试使用 sqlalchemist 创建一个数据库，以连接 snowflake 和 alembic 以迁移在 FastAPI 中创建的应用程序。我创建了一些模型，并且在雪花中创建这个模型时一切正常，例如：

create or replace TABLE PRICE_SERVICE.FP7.LOCATION (
    ID NUMBER(38,0) NOT NULL autoincrement,
    CREATED_AT TIMESTAMP_NTZ(9),
    UPDATED_AT TIMESTAMP_NTZ(9),
    ADDRESS VARCHAR(16777216),
    LATITUDE VARCHAR(16777216) NOT NULL,
    LONGITUDE VARCHAR(16777216) NOT NULL,
    unique (LATITUDE),
    unique (LONGITUDE),
    primary key (ID)
);

但是当我尝试为这个表创建一个新的对象时，我得到了：

sqlalchemy.orm.exc.FlushError: Instance <Location at 0x7fead79677c0> has a NULL identity key.  If this is an auto-generated value, check that the database table allows generation of new primary key values, and that the mapped Column object is configured to expect these generated values.  Ensure also that this flush() is not occurring at an inappropriate time, such as within a load() event.

我的模型是：

class Location(Base):
    id = Column(Integer, primary_key=True)
    address = Column(String)
    latitude = Column(String, unique=True, nullable=False)
    longitude = Column(String, unique=True, nullable=False)

    buildings = relationship("Building", back_populates="location")
    quotes = relationship("Quote", back_populates="location")
    binds = relationship("Bind", back_populates="location")

我正在尝试这样做：

def create_location(db: Session, data: Dict[str, Any]) -> Location:
    location = Location(
        address=data["address"],  # type: ignore
        latitude=data["lat"],  # type: ignore
        longitude=data["lng"],  # type: ignore
    )
    db.add(location)
    db.commit()
    
    return location

我也尝试使用：

id = Column(Integer, Sequence("id_seq"), primary_key=True)

但我得到了：

 sqlalchemy.exc.StatementError: (sqlalchemy.exc.ProgrammingError) (snowflake.connector.errors.ProgrammingError) 000904 (42000): SQL compilation error: error line 1 at position 7
backend_1  | invalid identifier 'ID_SEQ.NEXTVAL'

【问题讨论】：

标签： python sqlalchemy snowflake-cloud-data-platform fastapi

【解决方案1】：

您忘记在模型中定义 Sequence。当您在 Snowflake 中定义表创建时的序列值时，将在架构级别生成 Sequence。

from sqlalchemy import Column, Integer, Sequence
...


class Location(Base):
    id = Column(Integer, Sequence("Location_Id"), primary_key=True, 
        autoincrement=True)
    address = Column(String)
...

确保您的用户角色对该序列具有 usage 权限，这应该可以解决您为主键设置下一个值的问题。

一种帮助我处理表主键的方法是定义一个混合类，它使用 declared_attr 根据表名自动定义我的主键。

from sqlalchemy import Column, Integer, Sequence
from slqalchemy.ext.declarative import declared_attr


class SomeMixin(object):
    @declared_attr
    def record_id(cls):
        """
        Use table name to define pk
        """"
        return Column(
            f"{cls.__tablename__} Id",
            Integer(),
            primary_key=True, 
            autoincrement=True
        )

然后你将所说的 mixin 传递到你的模型中

from sqlalchemy import Column, Integer, String, Sequence
from wherever import SomeMixin

class Location(Base, SomeMixin):
    address = Column(String)
    ...

现在 Location.record_id 通过您在混入中定义的序列设置。

希望这有帮助

【讨论】：