我需要測驗一個 python 燒瓶應用程式，該應用程式使用 mySQL 使用 sqlalchemy 和 sqlite3 運行它的查詢。
ON DUPLICATE嘗試使用子句測驗 upsert 函式時遇到例外：
(sqlite3.OperationalError) near "DUPLICATE": syntax error
在對解決方案進行了簡短搜索后，我發現 sqlite 執行 upsert 查詢的正確語法是ON CONFLICT(id) DO UPDATE SET ...，我已經嘗試過，但 mySQL 無法識別此語法。
我能做些什么？如何進行 upsert 查詢，以便 sqlite3 和 mySQL 都能正確執行？
例子：
員工表：
| ID | 姓名 |
|---|---|
| 1 | 杰夫·貝索斯 |
| 2 | 比爾蓋茨 |
INSERT INTO employees(id,name)
VALUES(1, 'Donald Trump')
ON DUPLICATE KEY UPDATE name = VALUES(name);
應將表更新為：
| ID | 姓名 |
|---|---|
| 1 | 唐納德·特朗普 |
| 2 | 比爾蓋茨 |
提前致謝！
uj5u.com熱心網友回復：
如何進行 upsert 查詢，以便 sqlite3 和 mySQL 都能正確執行？
您可以通過嘗試 UPDATE 來獲得相同的結果，如果找不到匹配項，則執行 INSERT。以下代碼使用 SQLAlchemy Core 構造，它提供了進一步的保護，防止 MySQL 和 SQLite 之間的細微差異。例如，如果你的表有一個名為“order”的列，那么 SQLAlchemy 會為 MySQL 發出這個 DDL……
CREATE TABLE employees (
id INTEGER NOT NULL,
name VARCHAR(50),
`order` INTEGER,
PRIMARY KEY (id)
)
… 還有這個 SQLite 的 DDL
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')]
"""
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/485338.html
標籤：Python mysql sqlite sqlalchemy