dblocks.py 文件源码

def __enter__(self):
        """Obtain lock using pg_advisory_xact_lock()."""
        if not connection.in_atomic_block:
            raise DatabaseLockAttemptOutsideTransaction(self)
        with closing(connection.cursor()) as cursor:
            query = "SELECT %s(%%s, %%s)" % self.lock
            cursor.execute(query, self)
            if cursor.fetchone() == (False,):
                raise DatabaseLockNotHeld(self)