Insert twice with and without PK raises Duplicate primary key violates unique constraint

[meschac38700]

First Check

• I added a very descriptive title here.
• I used the GitHub search to find a similar question and didn't find it.
• I searched the SQLModel documentation, with the integrated search.
• I already searched in Google "How to X in SQLModel" and didn't find any information.
• I already read and followed all the tutorial in the docs and didn't find an answer.
• I already checked if it is not related to SQLModel but to Pydantic.
• I already checked if it is not related to SQLModel but to SQLAlchemy.

Commit to Help

• I commit to help with one of those options 👆

Example Code

from typing import Optional

from sqlmodel import Field, Session, SQLModel, create_engine


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: Optional[int] = None


hero_1 = Hero(id=1, name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(name="Deadpond", secret_name="Dive Wilson")

engine = create_engine("sqlite:///database.db")


SQLModel.metadata.create_all(engine)

with Session(engine) as session:
    session.add(hero_1)
    session.add(hero_2) # raise sqlalchemy.exc.IntegrityError: [...] DETAIL:  Key (id)=(1) already exists.
    session.commit()
    session.refresh(hero_1)
    session.refresh(hero_2)

Description

Hello, I am encountering an error while using SQLModel/SQLAlchemy.

I am using PostrgreSQL database and when I try to insert a row, after specifying the id (primary key) attribute,
the second insert (without specifying the id) fails due to a conflict over the unique constraint on the primary key.
The second insert attempts to insert the row with id 1 which was already taken with the first insert in this case.

This seems weird to me because sometimes I need to save data with the ID (like for my tests, initial data etc.).

And not being able to specify an ID is quite annoying.

Is the problem known? because I didn't find any open issues regarding this.

Operating System

Linux

Operating System Details

OS: Manjaro Linux x86_64
Kernel: Linux 6.6.60-1-lts

SQLModel Version

0.0.22

Python Version

Python 3.13.0

Additional Context

No response

[monchin]

What if you use sqlalchemy? It's better to clarify which this issue is related to.

[meschac38700]

@monchin Actually, I tried with sqlalchemy, no problem. the following code, using sqlalchemy's insert directly, no problem with auto_increment of the ID

from sqlalchemy import insert
from sqlmodel import Session, SQLModel, create_engine

engine = create_engine("sqlite:///database.db")


SQLModel.metadata.create_all(engine)

with Session(engine) as session:
    # Using sqlalchemy's `insert` function
    stmt = insert(Hero).values(id=1, name="username", secret_name="Full Username")
    stmt2 = insert(Hero).values(name="username1", secret_name="Full Username2")
    session.exec(stmt)
    session.exec(stmt2)

[meschac38700]

@monchin have you tried it with postgresql or mysql? I had errors with these databases. I will try your SQLite suggestion later.
Moreover these tests are executed in an empty database, if you have doubts, you always have the PR link of my project to test in an appropriate environment.

[meschac38700]

@monchin I tried your code with sqlite and it works even without having to drop the whole database but doesn't work when I use the postgres asynchronous engine with asyncpg

aiosqlite complete example

import asyncio
from typing import Optional

from sqlmodel import SQLModel, Field
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy_utils import database_exists, create_database, drop_database


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: Optional[int] = None


async def main():
   
    uri = "sqlite:///database.db"
    engine = create_engine("sqlite+aiosqlite:///database.db", echo=True)
    aengine = create_async_engine("sqlite+aiosqlite:///database.db", echo=True)
    """
    if database_exists(uri):
        drop_database(uri)
    create_database(uri)
    """
    async with aengine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.drop_all)
    async with aengine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

    hero_1 = Hero(id=1, name="Deadpond", secret_name="Dive Wilson")
    hero_2 = Hero(id=2, name="Deadpond1", secret_name="Dive Wilson1")
    hero_3 = Hero( name="Deadpond3", secret_name="Dive Wilson2")
    async with AsyncSession(aengine) as session:
        session.add(hero_1)
        session.add(hero_2)
        session.add(hero_3)
        await session.commit()
        await session.refresh(hero_1)
        await session.refresh(hero_2)
        await session.refresh(hero_3)

if __name__ == "__main__":
    asyncio.run(main())

[monchin]

Ok, I'll have a try on psql

[monchin]

Yes you're right, it crashed on psql. And I tried without sqlmodel with sync and async mode, with different drivers including psycopg, psycopg2, pg8000 and asyncpg, it crashed on every situation. So it can be concluded that this issue has no relationship with sqlmodel. I guess perhaps something different between psql and sqlite leads to this issue, but I have no idea to validate it.

Anyway, I suggest you not to explicitly specify the value of id. But if you must do that, you can open an issue on sqlalchemy and ask for some suggestions there.

Below is my sqlalchemy code:

import argparse
import asyncio
from typing import Optional

from sqlalchemy import String, create_engine
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import Mapped, Session, declarative_base, mapped_column
from sqlalchemy_utils import create_database, database_exists, drop_database

parser = argparse.ArgumentParser()
parser.add_argument("--asc", action="store_true", help="run with async mode")


my_psql_account = "..."
uri = f"postgresql+psycopg://{my_psql_account}@localhost:5432/asqltest"  # Faild on all the drivers
auri = f"postgresql+asyncpg://{my_psql_account}@localhost:5432/asqltest"
engine = create_engine(uri, echo=True)
aengine = create_async_engine(auri, echo=True)
if database_exists(uri):
    drop_database(uri)
create_database(uri)

Base = declarative_base()


class Hero(Base):
    __tablename__ = "hero"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    secret_name: Mapped[str] = mapped_column(String(30))
    age: Mapped[Optional[int]]


Base.metadata.create_all(engine)
hero_1 = Hero(id=1, name="Deadpond", secret_name="Dive Wilson")
hero_2 = Hero(id=2, name="Deadpond1", secret_name="Dive Wilson1")
hero_3 = Hero(name="Deadpond3", secret_name="Dive Wilson2")


async def amain():
    async with AsyncSession(aengine) as session:
        session.add(hero_1)
        session.add(hero_2)
        session.add(hero_3)
        await session.commit()
        await session.refresh(hero_1)
        await session.refresh(hero_2)
        await session.refresh(hero_3)


def main():
    with Session(engine) as session:
        session.add(hero_1)
        session.add(hero_2)
        session.add(hero_3)
        session.commit()
        session.refresh(hero_1)
        session.refresh(hero_2)
        session.refresh(hero_3)


if __name__ == "__main__":
    args = parser.parse_args()
    # Faild on both async and sync mode
    if args.asc:
        asyncio.run(amain())
    else:
        main()

[meschac38700]

@monchin Yeah Thx <3