Массовая вставка с SQLAlchemy ORM

SQLAlchemy представила, что в версии 1.0.0 :

Массовые операции – документы SQLAlchemy

С помощью этих операций теперь вы можете выполнять массовые вставки или обновления!

Например, вы можете:

 s = Session() objects = [ User(name="u1"), User(name="u2"), User(name="u3") ] s.bulk_save_objects(objects) 

Здесь будет вставлена ​​основная вставка.

Насколько я знаю, нет способа заставить ORM выдавать объемные вставки. Я считаю, что основная причина заключается в том, что SQLAlchemy должен отслеживать идентификацию каждого объекта (т. Е. Новые первичные ключи), а объемные вставки мешают этому. Например, если ваша таблица foo содержит столбец id и сопоставляется с классом Foo :

 x = Foo(bar=1) print x.id # None session.add(x) session.flush() # BEGIN # INSERT INTO foo (bar) VALUES(1) # COMMIT print x.id # 1 

Поскольку SQLAlchemy взял значение для x.id без выдачи другого запроса, мы можем заключить, что он получил значение непосредственно из INSERT . Если вам не нужен последующий доступ к созданным объектам через те же экземпляры, вы можете пропустить слой ORM для своей вставки:

 Foo.__table__.insert().execute([{'bar': 1}, {'bar': 2}, {'bar': 3}]) # INSERT INTO foo (bar) VALUES ((1,), (2,), (3,)) 

SQLAlchemy не может сопоставлять эти новые строки с любыми существующими объектами, поэтому вам придется запрашивать их заново для любых последующих операций.

Что касается устаревших данных, полезно помнить, что сеанс не имеет встроенного способа узнать, когда база данных изменена вне сеанса. Чтобы получить доступ к внешним измененным данным через существующие экземпляры, экземпляры должны быть отмечены как истекшие . Это происходит по умолчанию на session.commit() , но может быть сделано вручную, вызывая session.expire_all() или session.expire(instance) . Пример (SQL опущен):

 x = Foo(bar=1) session.add(x) session.commit() print x.bar # 1 foo.update().execute(bar=42) print x.bar # 1 session.expire(x) print x.bar # 42 

session.commit() истекает x , поэтому первый оператор печати неявно открывает новую транзакцию и повторно запрашивает атрибуты x . Если вы закомментируете первый оператор печати, вы заметите, что второй теперь получает правильное значение, потому что новый запрос не выдается до момента обновления.

Это имеет смысл с точки зрения изоляции транзакций – вы должны только получать внешние изменения между транзакциями. Если это вызывает у вас проблемы, я бы предложил уточнить или переосмыслить границы транзакций вашего приложения, а не сразу session.expire_all() к session.expire_all() .

В документах sqlalchemy есть отличная запись о производительности различных методов, которые могут использоваться для объемных вставок:

ORM в основном не предназначены для высокопроизводительных объемных вставок – в этом вся причина, по которой SQLAlchemy предлагает Core в дополнение к ORM как первоклассный компонент.

В случае использования быстрых вставных вставок система генерации и выполнения SQL, которую ORM строит сверху, является частью ядра. Используя эту систему напрямую, мы можем создать INSERT, который является конкурентным с использованием прямого API баз данных.

В качестве альтернативы ORM SQLAlchemy предлагает набор методов Bulk Operations, которые обеспечивают привязки к подразделам единицы рабочего процесса, чтобы испускать конструкторы INSERT и UPDATE на уровне ядра с небольшой степенью автоматизации на основе ORM.

Пример, приведенный ниже, иллюстрирует временные тесты для нескольких разных способов вставки строк, начиная от самого автоматического до минимума. С cPython 2.7 наблюдались время выполнения:

 classics-MacBook-Pro:sqlalchemy classic$ python test.py SQLAlchemy ORM: Total time for 100000 records 12.0471920967 secs SQLAlchemy ORM pk given: Total time for 100000 records 7.06283402443 secs SQLAlchemy ORM bulk_save_objects(): Total time for 100000 records 0.856323003769 secs SQLAlchemy Core: Total time for 100000 records 0.485800027847 secs sqlite3: Total time for 100000 records 0.487842082977 sec 

Автор сценария:

 import time import sqlite3 from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, create_engine from sqlalchemy.orm import scoped_session, sessionmaker Base = declarative_base() DBSession = scoped_session(sessionmaker()) engine = None class Customer(Base): __tablename__ = "customer" id = Column(Integer, primary_key=True) name = Column(String(255)) def init_sqlalchemy(dbname='sqlite:///sqlalchemy.db'): global engine engine = create_engine(dbname, echo=False) DBSession.remove() DBSession.configure(bind=engine, autoflush=False, expire_on_commit=False) Base.metadata.drop_all(engine) Base.metadata.create_all(engine) def test_sqlalchemy_orm(n=100000): init_sqlalchemy() t0 = time.time() for i in xrange(n): customer = Customer() customer.name = 'NAME ' + str(i) DBSession.add(customer) if i % 1000 == 0: DBSession.flush() DBSession.commit() print( "SQLAlchemy ORM: Total time for " + str(n) + " records " + str(time.time() - t0) + " secs") def test_sqlalchemy_orm_pk_given(n=100000): init_sqlalchemy() t0 = time.time() for i in xrange(n): customer = Customer(id=i+1, name="NAME " + str(i)) DBSession.add(customer) if i % 1000 == 0: DBSession.flush() DBSession.commit() print( "SQLAlchemy ORM pk given: Total time for " + str(n) + " records " + str(time.time() - t0) + " secs") def test_sqlalchemy_orm_bulk_insert(n=100000): init_sqlalchemy() t0 = time.time() n1 = n while n1 > 0: n1 = n1 - 10000 DBSession.bulk_insert_mappings( Customer, [ dict(name="NAME " + str(i)) for i in xrange(min(10000, n1)) ] ) DBSession.commit() print( "SQLAlchemy ORM bulk_save_objects(): Total time for " + str(n) + " records " + str(time.time() - t0) + " secs") def test_sqlalchemy_core(n=100000): init_sqlalchemy() t0 = time.time() engine.execute( Customer.__table__.insert(), [{"name": 'NAME ' + str(i)} for i in xrange(n)] ) print( "SQLAlchemy Core: Total time for " + str(n) + " records " + str(time.time() - t0) + " secs") def init_sqlite3(dbname): conn = sqlite3.connect(dbname) c = conn.cursor() c.execute("DROP TABLE IF EXISTS customer") c.execute( "CREATE TABLE customer (id INTEGER NOT NULL, " "name VARCHAR(255), PRIMARY KEY(id))") conn.commit() return conn def test_sqlite3(n=100000, dbname='sqlite3.db'): conn = init_sqlite3(dbname) c = conn.cursor() t0 = time.time() for i in xrange(n): row = ('NAME ' + str(i),) c.execute("INSERT INTO customer (name) VALUES (?)", row) conn.commit() print( "sqlite3: Total time for " + str(n) + " records " + str(time.time() - t0) + " sec") if __name__ == '__main__': test_sqlalchemy_orm(100000) test_sqlalchemy_orm_pk_given(100000) test_sqlalchemy_orm_bulk_insert(100000) test_sqlalchemy_core(100000) test_sqlite3(100000) 

Обычно я делаю это с помощью add_all .

 from app import session from models import User objects = [User(name="u1"), User(name="u2"), User(name="u3")] session.add_all(objects) session.commit() 

Прямая поддержка была добавлена ​​в SQLAlchemy с версии 0.8

В соответствии с документами , connection.execute(table.insert().values(data)) должен делать трюк. (Обратите внимание, что это не то же самое, что connection.execute(table.insert(), data) что приводит к executemany многих отдельных вставок строк при вызове executemany ). На чем угодно, кроме локального соединения, разница в производительности может быть огромной.

Это способ:

 values = [1, 2, 3] Foo.__table__.insert().execute([{'bar': x} for x in values]) 

Это будет выглядеть следующим образом:

 INSERT INTO `foo` (`bar`) VALUES (1), (2), (3) 

Ссылка: в SQLAlchemy FAQ включены эталоны для различных методов фиксации.

SQLAlchemy представила, что в версии 1.0.0 :

Массовые операции – документы SQLAlchemy

С помощью этих операций теперь вы можете выполнять массовые вставки или обновления!

Например (если вам нужны самые низкие накладные расходы для простых табличных INSERT), вы можете использовать Session.bulk_insert_mappings() :

 loadme = [ (1, 'a') , (2, 'b') , (3, 'c') ] dicts = [] for i in range(len(loadme)): dicts.append(dict(bar=loadme[i][0], fly=loadme[i][1])) s = Session() s.bulk_insert_mappings(Foo, dicts) s.commit() 

Или, если хотите, пропустите loadme кортежи и напишите словари непосредственно в dicts (но мне легче оставить всю информацию из данных и загрузить список словарей в цикле).

Ответ bulk_save_objects правильный, но одна проблема заключается в том, что bulk_save_objects по умолчанию не возвращает первичные ключи объектов, если это вас беспокоит. Установите return_defaults в True чтобы получить это поведение.

Документация находится здесь .

 foos = [Foo(bar='a',), Foo(bar='b'), Foo(bar='c')] session.bulk_save_objects(foos, return_defaults=True) for foo in foos: assert foo.id is not None session.commit()