how to create a auto-generated value to snowflake using sqlalchemist?

Question

I'm trying to create a db using sqlalchemist to connect with snowflake and alembic to migrations for an app created in FastAPI. I created some models and all works fine to create this one in snowflake for examples:

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)
);

but when I try to create a new obj to this table and I'm getting:

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.

my model is:

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")

and I'm trying to do this:

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

also I tried using:

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

but I got:

 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'

Answer 1

You forgot to define the Sequence in your model. When you define the Sequence value on table creation in Snowflake a Sequence is generated at the schema level.

from sqlalchemy import Column, Integer, Sequence
...


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

Make sure your user role has usage permission for that sequence and that should take care of your issue setting the next value for your primary key.

An approach that helps me with table primary keys is defining a mixin class that uses declared_attr to automatically define my primary keys based on the table name.

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
        )

Then you pass said mixin into your model

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

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

Now Location.record_id gets set through the sequence you defined in the mixin.

Hope this helped