Creating an autoincrement column via sqlalchemy

243 views Asked by At

I am using a Postgresql database via sqlalchemy orm. The table is really basic:

import sqlalchemy as sa
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column

class Network(Base):
    __tablename__ = 'networks'

    id: Mapped[int] = mapped_column(sa.Integer, primary_key=True)
    name: Mapped[str] = mapped_column(sa.String(60))
    type: Mapped[str] = mapped_column(sa.String(60))
    version: Mapped[int] = mapped_column(sa.Integer)

I would like to convert the version column into something which check if same name & same type row exists. If yes, then increase the previous version with 1.

So imagine my table already has following rows:

id    name    type    version
0     foo     big     1
1     bar     big     1

And I insert one more (foo, big) into the table then it should be automaticly:

id    name    type    version
0     foo     big     1
1     bar     big     1
2     foo     big     2

Is something like this possible? BTW, it should get also default version 1 if same name & same type does not exists.

1

There are 1 answers

0
hmn Falahi On

use before_insert event:

class Network(Base):
    __tablename__ = 'networks'

    id: Mapped[int] = mapped_column(sa.Integer, primary_key=True)
    name: Mapped[str] = mapped_column(sa.String(60))
    type: Mapped[str] = mapped_column(sa.String(60))
    version: Mapped[int] = mapped_column(sa.Integer)


def update_version(mapper, connection, target):
    session = Session.object_session(target)

    # Check if a row with the same name and type already exists
    existing_row = session.query(Network.version) \
        .filter(Network.name == target.name) \
        .filter(Network.type==target.type) \
        .order_by(Network.version.desc()) \
        .first()

    if existing_row:
        # If it exists, increase the version by 1
        target.version = existing_row.version + 1
    else:
        # If it doesn't exist, set the version to 1
        target.version = 1


event.listen(Network, 'before_insert', update_version)