Insertion en vrac avec SQLAlchemy ORM

SQLAlchemy introduit que dans la version 1.0.0 :

opérations en bloc - SQLAlchemy docs

avec ces opérations, vous pouvez maintenant faire des inserts en vrac ou des mises à jour!

Par exemple, vous pouvez faire:

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

ici, un encart en vrac sera fait.

autant que je sache, il n'y a aucun moyen d'obtenir de L'ORM qu'il émette des encarts en vrac. Je crois que la raison sous-jacente est que SQLAlchemy doit garder une trace de l'identité de chaque objet (c.-à-d., de nouvelles clés primaires), et inserts en vrac interfèrent avec cela. Par exemple, si votre tableau foo contient une colonne id et est mappé à une classe 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

depuis que SQLAlchemy a récupéré la valeur de x.id sans émettre une autre requête, nous pouvons en déduire qu'il a obtenu la valeur directement à partir de la INSERT . Si vous n'avez pas besoin d'un accès ultérieur aux objets créés via les instances mêmes , vous pouvez sauter la couche ORM pour votre insertion:

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

SQLAlchemy ne peut pas faire correspondre ces nouvelles lignes avec des objets existants, donc vous devrez les interroger à nouveau pour toute opération ultérieure.

en ce qui concerne les données périmées, il est utile de se rappeler que la session n'a pas de manière intégrée pour savoir quand la base de données est changée en dehors de la session. Afin d'accéder à des données modifiées par des instances existantes, les instances doivent porter la mention expirée . Cela se produit par défaut sur session.commit() , mais peut être fait manuellement en appelant session.expire_all() ou session.expire(instance) . Un exemple (SQL omis):

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() expire x , de sorte que la première déclaration imprimée implicitement ouvre une nouvelle transaction et ré-interroge les attributs de x . Si vous commentez la première instruction d'impression, vous remarquerez que la seconde affiche maintenant la bonne valeur, car la nouvelle requête n'est émise qu'après la mise à jour.

cela a du sens du point de vue de l'isolement transactionnel - vous ne devriez prendre des modifications externes entre les transactions. Si cela vous cause des problèmes, je suggère de clarifier ou de repenser votre demande les limites de transaction au lieu d'atteindre immédiatement session.expire_all() .

les docs sqlalchemy ont un grand writeup sur la performance de diverses techniques qui peuvent être utilisés pour inserts en vrac:

Les ORMs

ne sont essentiellement pas destinés à des inserts en vrac à hautes performances. - C'est la raison pour laquelle SQLAlchemy offre le noyau en plus de la ORM comme un composant de première classe.

pour le cas d'utilisation d'inserts en vrac rapide, la génération SQL et système d'exécution que l'ORM construit sur le dessus de fait partie de la Base. En utilisant ce système directement, nous pouvons produire un INSERT qui est en concurrence avec l'utilisation directe de L'API de base de données brute.

alternativement, le SQLAlchemy ORM offre la suite D'opérations en vrac de les méthodes qui fournissent des crochets dans les sous-sections de l'unité de travail processus afin d'émettre des inserts de niveau de base et de mettre à jour les constructions avec un petit degré d'automatisation basée sur L'ORM.

l'exemple ci-dessous illustre des essais basés sur le temps pour plusieurs méthodes d'insertion de lignes, allant du plus automatisé au moins. Avec le cPython 2.7, les durées d'exécution observées sont:

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

Script:

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)

je le fais habituellement en utilisant 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()

support Direct a été ajouté à SQLAlchemy à partir de la version 0.8

selon les docs , connection.execute(table.insert().values(data)) devrait faire l'affaire. (Notez qu'il s'agit de et non de comme connection.execute(table.insert(), data) , ce qui se traduit par de nombreuses insertions de lignes individuelles via un appel à executemany ). Sur tout autre chose qu'une connexion locale, la différence de performance peut être énorme.

SQLAlchemy introduit que dans la version 1.0.0 :

opérations en bloc - SQLAlchemy docs

avec ces opérations, vous pouvez maintenant faire des inserts en vrac ou des mises à jour!

par exemple (si vous voulez le plus bas au-dessus pour les INSERTs de table simples), vous pouvez utiliser 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()

ou, si vous voulez, sauter le loadme tuples et écrire les dictionnaires directement dans dicts (mais je trouve plus facile de laisser tout le wordiness de données et de charger une liste de dictionnaires dans une boucle).

la réponse de Piere est correcte mais un problème est que bulk_save_objects par défaut ne renvoie pas les clés primaires des objets, si cela vous concerne. Mettez return_defaults à True pour obtenir ce comportement.

la documentation est ici .

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

C'est une façon:

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

on insérera comme ceci:

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

référence: the SQLAlchemy FAQ inclut des benchmarks pour diverses méthodes de commit.

toutes les routes mènent à Rome , mais certains d'entre eux traverse les montagnes, nécessite des ferries, mais si vous voulez y arriver rapidement, il suffit de prendre l'autoroute.

dans ce cas, l'autoroute doit utiliser le execute_batch () caractéristique de psycopg2 . La documentation le dit le mieux:

la mise en œuvre actuelle de executemany() est (en utilisant une de bienfaisance euphémisme) n'est pas particulièrement performant. Ces fonctions peuvent être utilisées pour accélérer l'exécution répétée d'un énoncé par rapport à un ensemble de paramètres. En réduisant le nombre de roundtrips de serveur la performance peut être des ordres de grandeur mieux que d'utiliser executemany() .

dans mon propre test execute_batch() est environ deux fois plus rapide que executemany() , et donne l'option de configurer le page_size pour d'autres ajustements (si vous voulez presser les 2 à 3% de performances du pilote).

la même fonctionnalité peut facilement être activée si vous utilisez SQLAlchemy en paramétrant use_batch_mode=True comme paramètre lorsque vous instanciez le moteur avec create_engine()

la meilleure réponse que j'ai trouvée jusqu'à présent était dans la documentation de sqlalchemy:

http://docs.sqlalchemy.org/en/latest/faq/performance.html#i-m-inserting-400-000-rows-with-the-orm-and-it-s-really-slow

il existe un exemple complet de référence de solutions possibles.

comme indiqué dans la documentation:

bulk_save_objects n'est pas la meilleure solution mais les performances informatiques sont correct.

la deuxième meilleure mise en œuvre en termes de lisibilité je pense était avec le noyau SQLAlchemy:

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

le contexte de cette fonction est donné dans l'article de documentation.