CODEWITHSUNDEEP
pythonormsqlalchemydeclarative
Nemoden
Nemoden

Reputation: 9056

SQLAlchemy update if unique key exists

I've got a class:

class Tag(Base, TimestampMixin):
    """Tags"""
    __tablename__ = 'tags'
    __table_args__ = {'mysql_engine' : 'InnoDB', 'mysql_charset' : 'utf8' }

    id = Column(Integer(11), autoincrement = True, primary_key = True)
    tag = Column(String(32), nullable = False, unique = True)
    cnt = Column(Integer(11), index = True, nullable = False, default = 1)

    def __init__(self, tag):
        t = session.query(Tag).filter_by(tag=tag).first()
        if t:
            self.cnt = t.cnt+1
            self.tag = t.tag
        else:
            self.tag = tag

    def __repr__(self):
        return "<Tag('%s')>" % (self.tag, )

    def __unicode__(self):
        return "%s" % (self.tag, )

When adding tag:

tag = Tag('tag')
session.add(tag)
session.commit()

I want it to update existing tag.

Of course, I could've done this:

tag = session.query(Tag).filter_by(tag='tag').first()
if tag:
    tag.cnt++
else:
    tag = Tag('tag')
session.add(tag)
session.commit()

but, keeping such logic in Tag class seems to be more clear - possibly keeps me off of the shotgun surgery.

How do I get there? I'm pretty new to Python and SQLAlchemy, so any additional thoughts on my code will be appreciated.

Upvotes: 32

Views: 46268

Answers (3)

Ken
Ken

Reputation: 1461

This is how I implemented it with session:

def add_listing(row):
    """
    Add new listing to db if it doesn't exist, update it if it does
    """
    try:
        # Check if primary key exists already in table
        c_mlnum = row["ML #"]
        listing = db_session.query(Listings).filter(Listings.MLNum == c_mlnum)
        dup_listing = listing.first()
        if dup_listing:
            listing.update(
                {
                    v: row[k]
                    for k, v in listing_headers.items()
                    if k in row.index and not isnan(row[k]) and k != "ML #"
                }
            )
        else:
            record = Listings(
                **{
                    v: row[k]
                    for k, v in listing_headers.items()
                    if k in row.index and not isnan(row[k])
                }
            )
            db_session.add(record)  # Add current row
        db_session.commit()
    except Exception as e:
        # Rollback and print error
        db_session.rollback()
        print(e)

Upvotes: 1

vishes_shell
vishes_shell

Reputation: 23484

From version 1.2 SQLAlchemy will support on_duplicate_key_update for MySQL

There is also examples of how to use it:

from sqlalchemy.dialects.mysql import insert

insert_stmt = insert(my_table).values(
    id='some_existing_id',
    data='inserted value')

on_duplicate_key_stmt = insert_stmt.on_duplicate_key_update(
    data=insert_stmt.values.data,
    status='U'
)

conn.execute(on_duplicate_key_stmt)

From version 1.1 SQLAlchemy support on_conflict_do_update for PostgreSQL

Examples:

from sqlalchemy.dialects.postgresql import insert

insert_stmt = insert(my_table).values(
    id='some_existing_id',
    data='inserted value')

do_update_stmt = insert_stmt.on_conflict_do_update(
    constraint='pk_my_table',
    set_=dict(data='updated value')
)

conn.execute(do_update_stmt)

Upvotes: 20

Neel
Neel

Reputation: 21243

You can try this

def get_or_increase_tag(tag_name):
    tag = session.query(Tag).filter_by(tag=tag_name).first()
    if not tag:
       tag = Tag(tag_name)
    else:
       tag.cnt += 1
    return tag

You can check the link https://stackoverflow.com/search?q=Insert+on+duplicate+update+sqlalchemy

Upvotes: 18

Related Questions