Table temporaire SQLite à travers les threads

Question

J'utilise declarative_base de SQLAlchemy pour décrire mon schéma de données. Dans mon cas d'utilisation, je vais insérer une grande quantité de données dans la base de données chaque jour, puis utiliser et manipuler ces données en tant qu'objets métier après le fait.

Mon approche initiale consistait à configurer les tables en utilisant la méthodologie ORM, que la plupart de ma «logique métier» va interagir avec, cependant j'utilise le Core pour faire les insertions. Il y a quelques raisons à cela. Je dois d'abord me procurer les données sur un service Web en utilisant des centaines de requêtes individuelles. Chaque requête se traduira par des centaines/milliers de lignes de base de données. Avec un schéma très normalisé à travers de nombreuses tables et beaucoup ForeignKeys, je pensais juste que l'approche la plus simple serait la suivante:

1. Créer une table de mise en scène temporaire pour effectuer des insertions en vrac sous forme de données vient sur le webservice. Un thread sera dédié à la réception des résultats de la requête via un Queue et l'insertion dans la table.
2. Une fois que toutes les données ont été insérées dans la table temporaire, je vais exécuter plusieurs requêtes insert().from_select(...) pour traiter l'intégrité des données et tout mettre à la bonne place. Cela se fera à partir d'un thread séparé
3. Ceci est une fois un processus de jour et une fois le fait, le côté « logique métier » de mon application avec l'ORM sera disponible et faire son travail comme une application séparée

Mon problème est de maintenir les tables temporaires sur une seule connexion. Je semble perdre et ne peut pas comprendre pourquoi comme je l'ai suivi the approach in the docs

Les bits correspondants de code:

db maisons module moteur et une fonction auxiliaire pour assurer un soutien ForeignKey est sur:

engine = create_engine('sqlite:///%s' % SQLDB, poolclass=StaticPool) 

def get_connection(): 
    conn = engine.connect() 
    conn.execute('pragma foreign_keys=on') 
    return conn 

thread principal crée la table:

db.StagingTable.__table__.create(db.engine) 

Chaque thread de travail (Il y a deux, l'un des inserts faisant, un faire le travail INSERT INTO SELECT) utilise la fonction d'assistance pour obtenir une connexion:

self.cn = db.get_connection() 

Traceback partielle:

File "C:\FAST\Python266\lib\site-packages\sqlalchemy\engine\base.py", line 917, in _execute_context 
    context) 
    File "C:\FAST\Python266\lib\site-packages\sqlalchemy\engine\default.py", line 432, in do_executemany 
    cursor.executemany(statement, parameters) 
OperationalError: (OperationalError) no such table: staging u'INSERT INTO staging 

Qu'est-ce que je manque ici? J'étais sous l'impression qu'un StaticPool maintiendrait une connexion ouverte qui pourrait être passée à différents fils.

Answer 1

Vous devez faire au moins un Session.commit() pour écrire réellement quelque chose dans la base de données, y compris le DDL pour créer la table.

Le problème que vous avez vu était d'écrire INSERT dans une table qui n'avait pas encore été CREATE.