Допустим есть таблица
CREATE TABLE audio_file (
id INTEGER NOT NULL,
user_id INTEGER NOT NULL,
audio_list VARCHAR(255) NOT NULL,
PRIMARY KEY (id)
)
Заполняем тестовыми данными
(1, 1, 'audio1')
(2, 2, 'audio2')
(3, 3, 'audio3')
Создаем временную таблицу например
CREATE TABLE new_data (
user_id INTEGER NOT NULL,
audio_list VARCHAR(255) NOT NULL,
PRIMARY KEY (user_id)
)
Заполняем значениями словарей
data = [{'user_id': 2, 'audio_list': 'audio222'},
{'user_id': 3, 'audio_list': 'audio333'}]
Соединяем основную таблицу и временную и обновляем данные
update_tbl = (
update(audio_file).
where(audio_file.c.user_id == new_data.c.user_id).
values(audio_list=new_data.c.audio_list)
)
Получается
(1, 1, 'audio1')
(2, 2, 'audio222')
(3, 3, 'audio333')
Полный текст
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, insert, update,select
from sqlalchemy.inspection import inspect
# Создание движка базы данных
engine = create_engine('sqlite:///audio.db', echo=True)
# Создание метаданных
metadata = MetaData()
# Определение таблицы
audio_file = Table('audio_file', metadata,
Column('id', Integer, primary_key=True),
Column('user_id', Integer, nullable=False),
Column('audio_list', String(255), nullable=False),
)
# Создание таблицы
# Создаем инспектор для проверки существования таблицы
inspector = inspect(engine)
# Проверяем, существует ли таблица
if audio_file.name in inspector.get_table_names():
# Если таблица существует, удаляем и создаем ее
audio_file.drop(bind=engine)
audio_file.create(bind=engine)
else:
audio_file.create(bind=engine)
# Вставка данных в таблицу
with engine.connect() as connection:
with connection.begin():
data = [{'user_id': 1, 'audio_list': 'audio1'},
{'user_id': 2, 'audio_list': 'audio2'},
{'user_id': 3, 'audio_list': 'audio3'}]
connection.execute(insert(audio_file), data)
result=connection.execute(select(audio_file))
for el in result:
print(el)
# Создание временной таблицы с новыми данными
metadata = MetaData()
new_data = Table('new_data', metadata,
Column('user_id', Integer, primary_key=True),
Column('audio_list', String(255), nullable=False),
)
if new_data.name in inspector.get_table_names():
# Если таблицa существует, удаляем и создаем ее
new_data.drop(bind=engine)
new_data.create(bind=engine)
else:
new_data.create(bind=engine)
# Заполнение временной таблицы новыми данными
with engine.connect() as connection:
with connection.begin():
data = [{'user_id': 2, 'audio_list': 'audio222'},
{'user_id': 3, 'audio_list': 'audio333'}]
connection.execute(insert(new_data), data)
# Обновление таблицы audio_file данными из временной таблицы
with engine.connect() as connection:
with connection.begin():
update_tbl = (
update(audio_file).
where(audio_file.c.user_id == new_data.c.user_id).
values(audio_list=new_data.c.audio_list)
)
connection.execute(update_tbl)
result = connection.execute(select(audio_file))
for el in result:
print(el)
#Можно тут временную таблицу new_data удалить