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)
Answer from vishes_shell on Stack Overflow
Discussions

python - SQLAlchemy insert or update example - Stack Overflow
NOTE: different sql dialect may have different sql statement (sqlalchemy will handle that): ... There is no builtin function to do insert or update. However, you could query if a record exists first. More on stackoverflow.com
🌐 stackoverflow.com
if 'id' exists then update, else insert into MySQL database using Pandas dataframe
How's something like this? qry=db.query(ClassName).filter(ClassName.property == property): if qry.count(): # It exists already qry.property = newproperty db.session.merge(qry) else: # It doesn't exist yet new_object = ClassName(a=a, b=b) db.session.add(new_object) db.session.commit() More on reddit.com
🌐 r/learnpython
21
13
November 23, 2017
python - SQLAlchemy - performing a bulk upsert (if exists, update, else insert) in postgresql - Stack Overflow
I was under the impression that if the primary key already exists, the record would get updated. How can I upsert with Flask-SQLAlchemy based on primary key alone? Is there a simple solution? If there is not, I can always check for and delete any record with a matching id, and then insert the new ... More on stackoverflow.com
🌐 stackoverflow.com
python - sqlalchemy easy way to insert or update? - Stack Overflow
I have a sequence of new objects. They all look like similar to this: Foo(pk_col1=x, pk_col2=y, val='bar') Some of those are Foo that exist (i.e. only val differs from the row in the db) and should More on stackoverflow.com
🌐 stackoverflow.com
🌐
Towards Data Science
towardsdatascience.com › home › latest › the easiest way to upsert with sqlalchemy
The easiest way to UPSERT with SQLAlchemy | Towards Data Science
January 21, 2025 - As you see we use session.merge. This method analyzes our new_stock and checks which of the provided sku values are already present in the inventory table. If the sku exists it performs an update, otherwise it just inserts the new guitar.
🌐
SQLAlchemy
docs.sqlalchemy.org › en › 20 › core › dml.html
Insert, Updates, Deletes — SQLAlchemy 2.0 Documentation
The order of columns returned from this FROM clause should correspond to the order of columns sent as the names parameter; while this is not checked before passing along to the database, the database would normally raise an exception if these column lists don’t correspond. ... if True, non-server default values and SQL expressions as specified on Column objects (as documented in Column INSERT/UPDATE Defaults) not otherwise specified in the list of names will be rendered into the INSERT and SELECT statements, so that these values are also included in the data to be inserted.
🌐
SQLAlchemy
docs.sqlalchemy.org › en › 20 › orm › queryguide › dml.html
ORM-Enabled INSERT, UPDATE, and DELETE statements — SQLAlchemy 2.0 Documentation
The Session.execute() method, in addition to handling ORM-enabled Select objects, can also accommodate ORM-enabled Insert, Update and Delete objects, in various ways which are each used to INSERT, UPDATE, or DELETE many database rows at once. There is also dialect-specific support for ORM-enabled “upserts”, which are INSERT statements that automatically make use of UPDATE for rows that already exist.
Top answer
1 of 7
60

There is an upsert-esque operation in SQLAlchemy:

db.session.merge()

After I found this command, I was able to perform upserts, but it is worth mentioning that this operation is slow for a bulk "upsert".

The alternative is to get a list of the primary keys you would like to upsert, and query the database for any matching ids:

# Imagine that post1, post5, and post1000 are posts objects with ids 1, 5 and 1000 respectively
# The goal is to "upsert" these posts.
# we initialize a dict which maps id to the post object

my_new_posts = {1: post1, 5: post5, 1000: post1000} 

for each in posts.query.filter(posts.id.in_(my_new_posts.keys())).all():
    # Only merge those posts which already exist in the database
    db.session.merge(my_new_posts.pop(each.id))

# Only add those posts which did not exist in the database 
db.session.add_all(my_new_posts.values())

# Now we commit our modifications (merges) and inserts (adds) to the database!
db.session.commit()
2 of 7
53

You can leverage the on_conflict_do_update variant. A simple example would be the following:

from sqlalchemy.dialects.postgresql import insert

class Post(Base):
    """
    A simple class for demonstration
    """

    id = Column(Integer, primary_key=True)
    title = Column(Unicode)

# Prepare all the values that should be "upserted" to the DB
values = [
    {"id": 1, "title": "mytitle 1"},
    {"id": 2, "title": "mytitle 2"},
    {"id": 3, "title": "mytitle 3"},
    {"id": 4, "title": "mytitle 4"},
]

stmt = insert(Post).values(values)
stmt = stmt.on_conflict_do_update(
    # Let's use the constraint name which was visible in the original posts error msg
    constraint="post_pkey",

    # The columns that should be updated on conflict
    set_={
        "title": stmt.excluded.title
    }
)
session.execute(stmt)

See the Postgres docs for more details about ON CONFLICT DO UPDATE.

See the SQLAlchemy docs for more details about on_conflict_do_update.

Side-Note on duplicated column names

The above code uses the column names as dict keys both in the values list and the argument to set_. If the column-name is changed in the class-definition this needs to be changed everywhere or it will break. This can be avoided by accessing the column definitions, making the code a bit uglier, but more robust:

coldefs = Post.__table__.c

values = [
    {coldefs.id.name: 1, coldefs.title.name: "mytitlte 1"},
    ...
]

stmt = stmt.on_conflict_do_update(
    ...
    set_={
        coldefs.title.name: stmt.excluded.title
        ...
    }
)
Find elsewhere
🌐
Google Groups
groups.google.com › g › sqlalchemy › c › tVc19TUJQu8
Conditional insert in one transaction
This should be done at once by the database in order to avoid some conflicts or duplicates. An equivalent sql version of the transaction would be : IF EXISTS (SELECT * FROM Table1 WHERE Column1='SomeValue') INSERT INTO Table1 VALUES (...) Is there a way to do that using sqlalchemy ?
🌐
Readthedocs
docs-sqlalchemy.readthedocs.io › ko › latest › core › dml.html
Insert, Updates, Deletes — SQLAlchemy 1.1 Documentation
For INSERT and UPDATE, the values are the newly inserted/updated values. For DELETE, the values are those of the rows which were deleted. Upon execution, the values of the columns to be returned are made available via the result set and can be iterated using ResultProxy.fetchone() and similar.
🌐
Sling Academy
slingacademy.com › article › sqlalchemy-upsert-update-if-exists-insert-if-not
SQLAlchemy Upsert: Update if Exists, Insert if Not - Sling Academy
This tutorial explores how to execute an ‘upsert’ operation in SQLAlchemy, ensuring an efficient way to handle record creation or modification. An upsert is a database operation that inserts rows into a database table if they do not already exist, or updates them if they do.
🌐
Restack
restack.io › p › adding-columns-sqlalchemy-models-answer-insert-if-not-exists
Sqlalchemy Insert If Not Exists | Restackio
... Learn how to use SQLAlchemy's exists function to efficiently check for the existence of records while adding columns to your models. ... When working with SQLAlchemy, performing upserts—insert if not exists, else update—can be efficiently managed using the insert() construct along with ...
🌐
SQLAlchemy
docs.sqlalchemy.org › en › 14 › core › dml.html
Insert, Updates, Deletes — SQLAlchemy 1.4 Documentation
October 19, 2021 - ValuesBase.return_defaults() - an alternative method tailored towards efficient fetching of server-side defaults and triggers for single-row INSERTs or UPDATEs. INSERT…RETURNING - in the SQLAlchemy 1.4 / 2.0 Tutorial · method sqlalchemy.sql.expression.Update.where(*whereclause)¶ · inherited from the DMLWhereBase.where() method of DMLWhereBase · Return a new construct with the given expression(s) added to its WHERE clause, joined to the existing clause via AND, if any.
Top answer
1 of 2
1

Unfortunately I am not sure what is performance of merge, but regarding to this post (which I highly recommend) the best option would be to use upsert and then just read from the table

from sqlalchemy.dialects.postgresql import insert

insert_stmt = insert(my_table).values(id='id',data='data')
do_nothing_stmt = insert_stmt.on_conflict_do_nothing(index_elements=['id'])
db.query(table).get("id")
2 of 2
0

It is possible to do it one call. It wasn't easy because insert.values(**items) doesn't allow a where clause combined with NOT EXISTS. We need a subquery but then we also need literals for every column and the keys for the insert.Also the literal for datetime caused a insert issue (formatting, maybe just postgres). I am wondering if there is a better way for the date issue.

In below example item_dict is a dictionary with field: value entries. It is only inserted if the invoice_number AND invoice_line_number don't already exist.

    # Create literals for select. Datetime conversion was needed because insert failed without it due to formatting issues.
    literals = [literal(val) if key != 'invoice_date' else literal(val.strftime("%Y-%m-%d")) for key,val in item_dict.items() ]
    
    # NOT exists subquery providing the literals.
    insert_select = select(literals).where(
        ~exists(literal(1)).where(
            (performance_row_table.c.invoice_number == item_dict['invoice_number']) &
            (performance_row_table.c.invoice_line_number == item_dict['invoice_line_number'])
        )
    )

    keys = [key for key in item_dict.keys()]
    # Insert using the subquery
    insert_statement = performance_row_table.insert().from_select(keys, insert_select).returning(performance_row_table)

    compiled = insert_statement.compile(compile_kwargs={"literal_binds": True})
    print(compiled)
    insert_result = session.execute(insert_statement).fetchone()
🌐
GitHub
gist.github.com › timtadh › 7811458
How to compile an INSERT ... ON DUPLICATE KEY UPDATE with SQL Alchemy with support for a bulk insert. · GitHub
import sqlalchemy ## SQLAlchemy Insert or Update for sessions, w00t! class Upsert(sqlalchemy.sql.expression.Insert): pass from sqlalchemy.ext.compiler import compiles @compiles(Upsert, "mysql") def compile_upsert(insert_stmt, compiler, **kwargs): if insert_stmt._has_multi_parameters: keys = insert_stmt.parameters[0].keys() else: keys = insert_stmt.parameters.keys() pk = insert_stmt.table.primary_key auto = None if (len(pk.columns) == 1 and isinstance(pk.columns.values()[0].type, sqlalchemy.Integer) and pk.columns.values()[0].autoincrement): auto = pk.columns.keys()[0] if auto in keys: keys.rem
🌐
Reddit
reddit.com › r/flask › sqlite: add an entry if not exists
r/flask on Reddit: sqlite: add an entry if not exists
August 13, 2015 -
...
art = models.Articles(title = art_title, date = datum, description = words, url = adresse, status = 1)
db.session.add(art)
db.session.commit()

Hi there

My code adds my entries but duplicates evrything if i reload my page. I would like to have a condition IF NOT EXISTS.

Sounds very simple but cant find a solution :(

obi wan kenobi you're my only hope...

thanks

🌐
Reddit
reddit.com › r/learnpython › how to insert into my sqlite database or update the row if some " url " already exists in my scrapy app ?
r/learnpython on Reddit: How to Insert into my sqlite Database or Update the row if some " url " already exists in my Scrapy app ?
July 25, 2019 -

Hello, I am scraping a website with a pagination, inserting all the information into the datebase . I am using Scrapy pipelines in my scrapy app, sqlite3 database and sqlalchemy. Here are my codes and afterwards you will see my question.

items.py

import scrapy


class IndiSpiderItem(scrapy.Item):
    my_url = scrapy.Field()
    my_title = scrapy.Field()
    that_description = scrapy.Field()
    creator_name = scrapy.Field()
    backers_money = scrapy.Field()
    all_backers = scrapy.Field()
    left_days = scrapy.Field()
    tag_location = scrapy.Field()
    users_image = scrapy.Field()

models.py

import sqlalchemy
DeclarativeBase = declarative_base()
import datetime


def db_connect():
    conn = sqlalchemy.create_engine("sqlite:///startups_gogo")
    return conn

def create_table(engine):
    DeclarativeBase.metadata.create_all(engine)


class QuoteDB(DeclarativeBase):
    __tablename__ = "camp_table"

    id = Column(Integer, primary_key=True)
    url = Column('url', String(200))
    title = Column('title', Text())
    description = Column('description', Text())
    creators_name = Column('creators_name', String(120))
    backed_money = Column ('backed_money', String(20))
    backers = Column('backers', String(25))
    days_left = Column("days_left", String(15))
    tag_location = Column('tag_location', String(250))
    image = Column('image', String(250) )
    date = Column(DateTime)
    # date = Column('date', DateTime, default=datetime.datetime.utcnow, nullable=True)

pipelines.py

from sqlalchemy.orm import sessionmaker
from .models import QuoteDB, db_connect, create_table
import sqlalchemy

import datetime as dt
from datetime import date, timedelta


class IndiSpiderPipeline(object):
    def __init__(self):
        engine = sqlalchemy.create_engine("sqlite:///startup_gogo")
        create_table(engine)
        self.Session = sessionmaker(bind=engine)

    def process_item(self, item, spider):

        session = self.Session()
        quotedb = QuoteDB()
        quotedb.url = item['url']
        quotedb.description = item['description']
        quotedb.creators_name = item['creators_name']
        quotedb.backed_money = item['backed_money']
        quotedb.backers = item['backers']
        quotedb.days_left = item['days_left']
        quotedb.tag_location = item['tag_location']
        quotedb.image = item['image']
        quotedb.date = dt.datetime.now()
        try:
            session.add(quotedb)
            session.commit()
        except:
            session.rollback()
            raise
        finally:
            session.close()
        return item

Well, all the data goes into the database correctly, but when i will scrape the second time the website, i will need to see , if the url exists in db , i will need to update that row with that new scraped information..., if the url doesn't exist in db i will need to insert ,so i need to check it.

I know there is some idea called upsert in sqlalchemy, but i cannot find certain tutorials in the internet which might help me to do that. i know there are some built in methods that sqlalchemy has for mysql and postgresql like

from sqlalchemy.dialects.postgresql import insert

But sqlite doesn't have such system.

How to write " Update if exists the 'url' , else insert into db".