1

我正在尝试编写一个 Python + SQLAlchemy 应用程序，它使用一个集中的 sqlite 数据库来存储有关各种文件的信息。我使用校验和来验证每个文件和/或文件版本在数据库中只存储一次，所以我UniqueConstraint在校验和列上有一个。

我将文件信息存储在DbFile要对其进行进一步操作的对象中。我创建了一个名为的静态方法New()，它将返回一个DbFile对象。如果文件已经存在于数据库中，它将返回该对象，否则将创建一个新对象。

 import md5
 # Not shown: sqlalchemy imports

 Base = declarative_base()
 engine = create_engine('sqlite:///test.db')
 Session = sessionmaker(bind=engine)
 session = Session()

 class DbFile(Base):
     id = Column(Integer, primary_key=True)
     filename = Column(String, nullable=False)
     checksum = Column(String, nullable=False)

     __table_args__ = (UniqueConstraint('checksum'),)

     @staticmethod
     def New(filename):
          file_md5 = md5.md5(open(filename).read())
          checksum = file_md5.hexdigest()

     q = session.query(Files).filter(Files.checksum == checksum)
     if q.count() == 1:
         print "Loading existing file object from database"
         return q.one()

     dbfile = DbFile(filename=filename, checksum=checksum)

在加载了一些文件后，我做了一个session.commit(). 如果只有一个进程正在访问数据库，这很好用，但是如果我启动多个 python 进程（从命令行），其中一个总是会中止并sqlalchemy.exc.IntegrityError抱怨校验和违反了唯一键约束。在检查存在和写入数据库之间显然存在竞争条件，但我还没有找到防止它的好方法。

我试图捕获错误，session.commit()但有时错误是由q.count()语句触发的。有没有一种干净的方法可以做到这一点？

4

0 回答 0