I need to test a python flask app that uses mySQL to run its' queries using sqlalchemy, with sqlite3.

I've encountered an exception when trying to test an upsert function using an ON DUPLICATE clause:

(sqlite3.OperationalError) near "DUPLICATE": syntax error

After a brief search for a solution, I've found that the correct syntax for sqlite to execute upsert queries is ON CONFLICT(id) DO UPDATE SET ..., I've tried it but mySQL doesn't recognize this syntax.

What can I do? How can I do an upsert query so sqlite3 and mySQL will both execute it properly?

Example:

employees table:

INSERT INTO employees(id,name)
VALUES(1, 'Donald Trump')
ON DUPLICATE KEY UPDATE name = VALUES(name);

Should update the table to be:

Thanks in advance!

CodePudding user response：

How can I do an upsert query so sqlite3 and mySQL will both execute it properly?

You can achieve the same result by attempting an UPDATE, and if no match is found then do an INSERT. The following code uses SQLAlchemy Core constructs, which provide further protection from the subtle differences between MySQL and SQLite . For example, if your table had a column named "order" then SQLAlchemy would emit this DDL for MySQL …

CREATE TABLE employees (
    id INTEGER NOT NULL, 
    name VARCHAR(50), 
    `order` INTEGER, 
    PRIMARY KEY (id)
)

… and this DDL for SQLite

CREATE TABLE employees (
    id INTEGER NOT NULL, 
    name VARCHAR(50), 
    "order" INTEGER, 
    PRIMARY KEY (id)
)

import logging

import sqlalchemy as sa

# pick one
connection_url = "mysql mysqldb://scott:tiger@localhost:3307/mydb"
# connection_url = "sqlite://"

engine = sa.create_engine(connection_url)


def _dump_table():
    with engine.begin() as conn:
        print(conn.exec_driver_sql("SELECT * FROM employees").all())


def _setup_example():
    employees = sa.Table(
        "employees",
        sa.MetaData(),
        sa.Column("id", sa.Integer, primary_key=True, autoincrement=False),
        sa.Column("name", sa.String(50)),
    )
    employees.drop(engine, checkfirst=True)
    employees.create(engine)

    # create initial example data
    with engine.begin() as conn:
        conn.execute(
            employees.insert(),
            [{"id": 1, "name": "Jeff Bezos"}, {"id": 2, "name": "Bill Gates"}],
        )


def upsert_employee(id_, name):
    employees = sa.Table("employees", sa.MetaData(), autoload_with=engine)
    with engine.begin() as conn:
        result = conn.execute(
            employees.update().where(employees.c.id == id_), {"name": name}
        )
        logging.debug(f" {result.rowcount} row(s) updated.")
        if result.rowcount == 0:
            result = conn.execute(
                employees.insert(), {"id": id_, "name": name}
            )
            logging.debug(f" {result.rowcount} row(s) inserted.")


if __name__ == "__main__":
    logging.basicConfig(level=logging.DEBUG)
    _setup_example()
    _dump_table()
    """
    [(1, 'Jeff Bezos'), (2, 'Bill Gates')]
    """
    upsert_employee(3, "Donald Trump")
    """
    DEBUG:root: 0 row(s) updated.
    DEBUG:root: 1 row(s) inserted.
    """
    _dump_table()
    """
    [(1, 'Jeff Bezos'), (2, 'Bill Gates'), (3, 'Donald Trump')]
    """
    upsert_employee(1, "Elon Musk")
    """
    DEBUG:root: 1 row(s) updated.
    """
    _dump_table()
    """
    [(1, 'Elon Musk'), (2, 'Bill Gates'), (3, 'Donald Trump')]
    """