Method read_pid

celery/platforms.py:163–176

Read and return the current pid.

(self)

    __exit__ = release

    def read_pid(self):
        """Read and return the current pid."""
        with ignore_errno('ENOENT'):
            with open(self.path) as fh:
                line = fh.readline()
                if line.strip() == line:  # must contain '\n'
                    raise ValueError(
                        f'Partial or invalid pidfile {self.path}')

                try:
                    return int(line.strip())
                except ValueError:
                    raise ValueError(
                        f'pidfile {self.path} contents invalid.')

    def remove(self):
        """Remove the lock."""