Wstawianie zbiorcze z SQLAlchemy ORM

Question

Czy istnieje sposób, aby SQLAlchemy zrobić wkładkę luzem zamiast wstawiania każdego pojedynczego obiektu. tj

robi:

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

zamiast:

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

Właśnie przekształcony kodu do korzystania sqlalchemy zamiast surowego sql i mimo że jest teraz dużo ładniejszy z nim pracować wydaje się być teraz wolniejszy (do 10 razy), zastanawiam się, czy to jest powód.

Może bym mógł poprawić sytuację, wykorzystując sesje bardziej efektywnie. W tej chwili mam autoCommit=False i robię session.commit() po dodaniu kilku rzeczy. Chociaż wydaje się, że dane zanikają, jeśli DB zostanie zmieniony gdzie indziej, nawet jeśli wykonuję nowe zapytanie, nadal otrzymuję stare wyniki?

Dzięki za pomoc!

Answer 1

O ile mi wiadomo, nie ma sposobu, aby ORM wydał inserty luzem. Uważam, że podstawową przyczyną jest to, że SQLAlchemy musi śledzić tożsamość każdego obiektu (tj. Nowe klucze podstawowe), a wkładki zbiorcze kolidują z tym. Na przykład, zakładając swoją tabelę foo zawiera kolumnę id i jest odwzorowany na Foo Klasa:

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 

Od SQLAlchemy podniósł wartość dla x.id bez wydawania kolejne zapytanie, możemy wywnioskować, że to ma wartość bezpośrednio z INSERT oświadczenie. Jeśli nie potrzebują kolejnego dostęp do tworzonych obiektów poprzez samych przypadkach można pominąć warstwę ORM dla wkładki:

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

SQLAlchemy nie może się równać z tych nowych wierszy z istniejących obiektów, tak będziesz musiał przesłać nowe zapytanie do kolejnych operacji.

Jeśli chodzi o nieaktualne dane, warto pamiętać, że sesja nie ma wbudowanego sposobu sprawdzania, czy baza danych została zmieniona poza sesją. Aby uzyskać dostęp do danych zmodyfikowanych zewnętrznie za pośrednictwem istniejących instancji, wystąpienia muszą być oznaczone jako wygasły. Dzieje się tak domyślnie na session.commit(), ale można to zrobić ręcznie, dzwoniąc pod numer session.expire_all() lub session.expire(instance). Przykładem (SQL pominięta):

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() wygasa x, więc pierwsza instrukcja print domyślnie otwiera nową transakcję i ponownie zapytań x „s atrybuty. Jeśli skomentujesz pierwszą instrukcję drukowania, zauważysz, że druga z nich odbiera teraz poprawną wartość, ponieważ nowe zapytanie nie zostanie wysłane dopiero po aktualizacji.

Ma to sens z punktu widzenia izolacji transakcyjnej - należy wybierać tylko zewnętrzne modyfikacje między transakcjami. Jeśli powoduje to problemy, sugeruję wyjaśnienie lub ponowne przemyślenie granic transakcji, zamiast natychmiastowego osiągnięcia wartości session.expire_all().

Answer 2

Bezpośrednia pomoc została dodana do SQLAlchemy od wersji 0.8

Zgodnie z docs, connection.execute(table.insert().values(data)) powinien zrobić lewy. (Należy zauważyć, że jest to , a nie tak samo jak connection.execute(table.insert(), data), co powoduje wiele pojedynczych wstawień wierszy za pośrednictwem wywołania executemany). Na poziomie innym niż połączenie lokalne różnica w wydajności może być ogromna.

Answer 3

Jest to sposób:

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

To będzie wstawić tak:

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

Odniesienie: SQLAlchemy FAQ zawiera odniesienia do różnych popełnić metod.

Answer 4

SQLAlchemy że wprowadzone w wersji 1.0.0:

Bulk operations - SQLAlchemy docs

z tymi operacjami, można teraz zrobić wkładki lub aktualizacji zbiorczych!

Na przykład, można zrobić:

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

Tutaj luzem wkładka zostanie wykonany.

Answer 5

docs sqlalchemy mają wielki writeup na wykonanie różnych technik, które można wykorzystać do wkładek masowych:

ORMs nie są zasadniczo przeznaczone do wysokowydajnych wkładek masowych - to jest cały powód SQLAlchemy oferuje Rdzeń oprócz ORM jako pierwszorzędnej części.

W przypadku użycia szybkich insertów luzem, system wykonawczy SQL i , na którym zbudowany jest ORM, jest częścią Rdzenia. Korzystając z tego systemu bezpośrednio, możemy wyprodukować WSTAW, który jest konkurencyjny z bezpośredniego korzystania z API bazy danych.

Alternatywnie SQLAlchemy ORM oferuje nasypowa Operacje zestaw metod, które zapewniają haki w podrozdziałach jednostki roboczej proces w celu emitowania INSERT Rdzeń poziomie i UPDATE konstruktów z mały stopień ORM oparte automatyzacja.

Poniższy przykład ilustruje testy oparte na czasie dla różnych metod wstawiania wierszy , przechodząc od najbardziej zautomatyzowanego do najmniejszego. Z CPython 2.7, czasy pracy zauważył:

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 

Scenariusz:

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) 

Answer 6

SQLAlchemy wprowadzono że w wersji 1.0.0:

Bulk operations - SQLAlchemy docs

z tymi operacjami, można teraz zrobić zbiorcze wstawki lub aktualizacje!

Na przykład (jeśli chcesz najmniejszy narzut dla prostych INSERT tabeli), można użyć 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() 

Albo, jeśli chcesz, pomiń loadme krotki i pisać słowników bezpośrednio do dicts (ale Uważam, że łatwiej jest zostawić wszystkie dane z danych i załadować listę słowników w pętli).

Answer 7

Odpowiedź Piere'a jest prawidłowa, ale jedną z kwestii jest domyślne, że domyślnie bulk_save_objects nie zwraca kluczy podstawowych obiektów, jeśli ma to znaczenie dla Ciebie. Ustaw return_defaults na True, aby uzyskać to zachowanie.

Dokumentacja to here.

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() 

Answer 8

Zwykle robię to za pomocą 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()