Method _init_db

ci/util/lock_database.py:47–70

Initialize database schema.

(self)

        return conn

    def _init_db(self) -> None:
        """Initialize database schema."""
        conn = self._get_connection()
        try:
            cursor = conn.cursor()
            cursor.execute(
                """
                CREATE TABLE IF NOT EXISTS lock_holders (
                    lock_name TEXT NOT NULL,
                    owner_pid INTEGER NOT NULL,
                    lock_mode TEXT NOT NULL CHECK(lock_mode IN ('read', 'write')),
                    operation TEXT NOT NULL DEFAULT 'lock',
                    hostname TEXT NOT NULL DEFAULT '',
                    acquired_at REAL NOT NULL,
                    PRIMARY KEY (lock_name, owner_pid)
                )
                """
            )
            cursor.execute(
                "CREATE INDEX IF NOT EXISTS idx_lock_name ON lock_holders(lock_name)"
            )
            conn.commit()
        finally:
            conn.close()

    def try_acquire(
        self,