小编Nik*_*res的帖子

我正在使用Flask-SQLAlchemy来做一个相当大的60k行的批量插入.我在这张桌子上也有多对多的关系,所以我不能用db.engine.execute它.在插入之前,我需要在数据库中找到类似的项目,如果找到重复的项目,则将插入更改为更新.

我可以事先做这个检查,然后通过插件进行批量插入db.engine.execute,但是在插入时我需要该行的主键.

目前,我做了db.session.add(),并db.session.commit()在每次插入,我也得到每秒进账3-4插入.

我运行了一个分析器,看看瓶颈在哪里,似乎db.session.commit()占用了60%的时间.

有没有什么方法可以让我更快地完成这个操作,可能是通过分组提交,但是哪个会给我主键？

这就是我的模型:

class Item(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(1024), nullable=True)
    created = db.Column(db.DateTime())
    tags_relationship = db.relationship('Tag', secondary=tags, backref=db.backref('items', lazy='dynamic'))
    tags = association_proxy('tags_relationship', 'text')

class Tag(db.Model):
id = db.Column(db.Integer, primary_key=True)
text = db.Column(db.String(255))

我的插入操作是这样运行的:

for item in items:
    if duplicate:
        update_existing_item
    else:
        x = Item()
        x.title = "string"
        x.created = datetime.datetime.utcnow()
        for tag in tags:
            if not tag_already_exists:
                y = Tag()
                y.text = "tagtext"
                x.tags_relationship.append(y) …

我正在使用Flask-SQLAlchemy,我需要创建一个没有自动刷新操作的会话.但是,使用Flask-SQLAlchemy创建的默认范围会话db.session已启用自动刷新功能.

我正在为100k行进行批量更新,并且自动刷新会导致严重的性能问题.

我尝试使用以下命令创建一个关闭自动刷新的新会话:

from flask_sqlalchemy import SignallingSession
db_session = SignallingSession(db, autoflush=False)

然后我尝试使用以下方法创建一个新行:

from flask_sqlalchemy import SQLAlchemy, BaseQuery

db = SQLAlchemy(app)

class Tool(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(1024), nullable=True)

add_tool = Tool()
add_tool.title = title

db_session.add(add_tool)

但是,这样做会引发以下异常:

Traceback (most recent call last):
  File "app", line 182, in <module>
called for last item
    import_tools('x.json', 1)
  File "app", line 173, in import_tools
    x.add_to_database()
  File "app", line 126, in add_to_database
    title=metadata['title'], commit=True)
  File "app", line 150, in transfer
    self.db_session.add(import_tool)
  File …

我一直熟悉Python中的unittest库,我写了几个unitest.TestCase类似于这个的:

class TestOne(unittest.TestCase):
    def setUp(self):
        pass

    def first_test(self):
        self.assertEqual('a', 'b')

    def second_test(self):
        self.assertEqual('a', 'b')

    def third_test(self):
        self.assertEqual('a', 'b')

    def tearDown(self):
        pass


class TestTwo(unittest.TestCase):
    def setUp(self):
        pass

    def first_test(self):
        self.assertEqual('a', 'b')

    def second_test(self):
        self.assertEqual('a', 'b')

    def third_test(self):
        self.assertEqual('a', 'b')

    def tearDown(self):
        pass


class TestThree(unittest.TestCase):
    def setUp(self):
        pass

    def first_test(self):
        self.assertEqual('a', 'b')

    def second_test(self):
        self.assertEqual('a', 'b')

    def third_test(self):
        self.assertEqual('a', 'b')

    def tearDown(self):
        pass

现在这不是我的代码完全如此,但这是遵循的基本结构.

我知道如果我想执行一个TestCase,我可以这样做:

suite = unittest.TestLoader().loadTestsFromTestCase(TestOne)
unittest.TextTestRunner(verbosity=2).run(suite)

但是,我一直在努力让多个TestCases同时运行.

我试过这样做:

suite = unittest.TestSuite()
suite.addTest(TestOne())
unittest.TextTestRunner(verbosity=2).run(suite) …