多线程上的sqlalchemy连接池

Question

首先我创建一个简单的表

import threading
from sqlalchemy import create_engine
from sqlalchemy import Column, Integer, Float, String, Date, DateTime, Boolean, select
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.pool import SingletonThreadPool
from sqlalchemy.orm import scoped_session
from sqlalchemy.orm import sessionmaker

inven_schema = {'__tablename__' : 'inventory',
            'item_no'       : Column(Integer, primary_key=True, autoincrement=True),
            'desc'          : Column(String(255), nullable=False),
            'volume'        : Column(Integer, nullable=False)
            }


Base = declarative_base()

# Dynamically create Inventory for ORM
Inventory = type('Inventory', (Base,), inven_schema)

some_inventory = [{'item_no' : 0, 'desc' : 'toy crane', 'volume' : 12},
              {'item_no' : 1, 'desc' : 'puddle jumper', 'volume' : 2},
              {'item_no' : 2, 'desc' : 'pet snake', 'volume' : 1},
              {'item_no' : 3, 'desc' : 'bowling ball', 'volume' : 4},
              {'item_no' : 4, 'desc' : 'spinning top', 'volume' : 3},
              {'item_no' : 5, 'desc' : 'pumpkin', 'volume' : 2}]

但是如果会话和插入在不同的线程上，我就不能让它工作。这是我的尝试，它抛出：

^{pr2}$

我尝试过在线程调用中创建会话，以及其他一些变体。我应该如何修改以下方法？在

def insert_inventory(inventory, session):
    inventory_ = [Inventory(**inven) for inven in inventory]
    session.add_all(inventory_)
    session.commit()

def results(session):
    q = select([Inventory.item_no]).distinct(Inventory.item_no)
    distinct_items = session.execute(q).fetchall()
    print('{!r}'.format(distinct_items))

def demo2():
    engine = create_engine('sqlite://')
    conn = engine.connect()
    Base.metadata.create_all(conn.engine)

    session_factory = sessionmaker(bind=conn.engine)

    Session = scoped_session(session_factory)
    session = Session()

    thread_0 = threading.Thread(target=insert_inventory, args=(some_inventory, session))
    thread_0.start()

    session.close()

    results(session)

Answer 1

我无法add()或add_all()级别。如果engine()是用其他参数创建的，则以下操作有效。在

我相信它是序列化对sqlite3db表的调用，这对sqlite来说并不理想，但却是必需的。在

engine = create_engine('sqlite://', connect_args={'check_same_thread' : False})
conn = engine.connect()

metadata = MetaData(engine)
table = Table('inventory',
              metadata,
              Column('item_no', Integer, primary_key=True, autoincrement=True),
              Column('desc', String(255), nullable=False),
              Column('volume', Integer, nullable=False)
              )

metadata.create_all()

some_inventory = [{'item_no' : 0, 'desc' : 'toy crane', 'volume' : 12},
                  {'item_no' : 1, 'desc' : 'puddle jumper', 'volume' : 2},
                  {'item_no' : 2, 'desc' : 'pet snake', 'volume' : 1},
                  {'item_no' : 3, 'desc' : 'bowling ball', 'volume' : 4},
                  {'item_no' : 4, 'desc' : 'spinning top', 'volume' : 3},
                  {'item_no' : 5, 'desc' : 'pumpkin', 'volume' : 2}]


thread_0 = threading.Thread(target=insert_inventory_table, args=(conn, table, some_inventory[0:3]))
thread_1 = threading.Thread(target=insert_inventory_table, args=(conn, table, some_inventory[3:]))

thread_0.start()
thread_1.start()