代码之家 › 专栏 › 技术社区 › Charles Pehlivanian

多写线程上的Sqlalchemy

python-multithreading sqlalchemy python-3.x python

Charles Pehlivanian · 技术社区 · 8 年前

以下代码适用于Python 3.6+,而不是Python 3.4.3,不确定在哪个版本失败。为什么会这样?我的印象是,sqlalchemy可能会将多个读写器隐藏在调用序列化器之后,从而处理基于文件的数据库的多个读写器。无论如何,这表明我没有正确处理这个问题-如何在版本中插入多个线程,或主线程外的一个线程<3.6?

我在sqlalchemy尝试过这个 session() 数量 sqlalchemy connection pool on multiple threads

但它只能和引擎一起工作,现在我只能在3.6版本中找到。

def insert_inventory_table(conn, table, inventory):
    conn.execute(table.insert(), inventory)    

def results_table(conn, table):
    q = select([table])
    data = conn.execute(q).fetchall()
    print('{!r}'.format(data))


def main_0():
    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' : 'drone', 'volume' : 12},
                      {'item_no' : 1, 'desc' : 'puddle jumper', 'volume' : 2},
                      {'item_no' : 2, 'desc' : 'pet monkey', 'volume' : 1},
                      {'item_no' : 3, 'desc' : 'bowling ball', 'volume' : 4},
                      {'item_no' : 4, 'desc' : 'electric guitar', 'volume' : 3},
                      {'item_no' : 5, 'desc' : 'bookends', '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()

    return conn, table


if __name__ == '__main__':

    conn,table = main_0()
    results_table(conn, table)

谢谢

2 回复 | 直到 8 年前

Kf H. 5 年前

如前所述,您必须使用作用域会话,因此,不要使用conn:

from sqlalchemy.orm import sessionmaker, scoped_session

db_session = scoped_session(sessionmaker(autocommit=False,
                                         autoflush=False,
                                         bind=engine))

然后,每当你在线程中连接到数据库时,记住打开和关闭会话,你不希望会话被线程共享,因为这会导致不好的事情,所以:

def worker(sth):
     session = db_session()
     "do sth very important"
     res = session.query(YourModel).filter(YourModel.sth = sth).first()
     print(res)
     session.close()

然后,您可以在多线程操作中使用这些worker。我希望这有帮助。

Kf H. 5 年前

让我在上面的答案中添加一些内容(因为我无法再编辑它)。使用作用域会话,您可以调用其他与db连接的函数,并且不必将会话作为参数添加到这些函数中,因为在同一线程中创建的下一个会话都是同一个会话。因此:

def worker(sth):
     session = db_session()
     "do sth very important"
     res = session.query(YourModel).filter(YourModel.sth == sth).first()
     your_very_importan_function(sth)

     print(res)
     db_session.remove()

def your_very_important_function(sth):
     session = db_session()    # this session will be the same as the session created in the worker function, and in the worker function, it is different for every thread.
    "do sth even more important"
    res = session.query(YourOtherModel).filter(YourOtherModel.sth == sth).first()

最后,您应该删除会话,而不仅仅是关闭它。 https://docs.sqlalchemy.org/en/13/orm/contextual.html