python - (Flask-)SQLAlchemy 主键问题可能是由于隐式事务

Question

在使用 Flask-SQLAlchemy 的项目中，我遇到了一些间歇性错误，我认为这可能是由于未明确使用事务。

我有这两个模型类，一个用于位置，另一个用于闭包：

class Location(db.Model):
    id = sa.Column(sa.Integer, primary_key=True)
    name = sa.Column(sa.String)
    code = sa.Column(sa.String, unique=True)


class LocationPath(db.Model):
    ancestor_id = sa.Column(sa.Integer, sa.ForeignKey('location.id'), nullable=False, primary_key=True)
    descendant_id = sa.Column(sa.Integer, sa.ForeignKey('location.id'), nullable=False, primary_key=True)
    depth = sa.Column(sa.Integer, default=0, nullable=False)

在后台进程中，我做了很多插入，所以我绕过 ORM 来使用 Core：

location_table = Location.__table__
location_path_table = LocationPath.__table__

statement = select([location_table.c.id]).where(code == code)
result = db.session.get_bind().execute(statement)
location_id = result.first()

if location_id is None:
    statement = location_table.insert().values(**kwargs)
    result = db.session.get_bind().execute(statement)
    new_id = result.inserted_primary_key[0]
    result.close()
else:
    new_id = location_id
# save new_id as an ancestor_id or a descendant_id

path = LocationPath.query.filter_by(
    ancestor_id=ancestor_id,
    descendant_id=descendant_id
).first()
if path is None:
    statement = location_path_table.insert().values(
        ancestor_id=ancestor_id,
        descendant_id=descendant_id,
        depth=depth)
    # the line below intermittently generates either of two errors:
    # - the inserted primary key (ancestor/descendant) does not exist
    # - a duplicate key error where the path already exists
    result = db.session.get_bind().execute(statement)

这导致我相当头疼，因为我从 a或 an获取ancestor_idor ，并且我还在尝试插入路径之前查询数据库以查看路径是否存在。descendant_idselectinsert

编辑：上面的代码循环运行。

标签： pythonsqlalchemyflask-sqlalchemy