docs: def acquire(self, key: str, blocking: bool=True, timeout: float=None, metadata: Any=None,
                  on_before_lock: LockEventListener=lambda key: None,
                  on_lock_already_locked: LockEventListener=lambda key: None,
                  lock_poll_interval_generator: Callable[[int], float]=DEFAULT_LOCK_POLL_INTERVAL_GENERATOR) \
              -> Optional[ConnectedConsulLockInformation]:
          """
          Acquires a Consul lock.
          :param key: the lock key
          :param blocking: whether to block and wait for the lock
          :param timeout: timeout in seconds
          :param metadata: metadata to add to the lock information. Must be parsable by default JSON encode/decoder
          :param on_before_lock: event listener to be called before attempt to acquire lock
          :param on_lock_already_locked: event listener to be called when an attempt to acquire a lock has failed as the
          lock is already locked
          :param lock_poll_interval_generator: generator of the interval between Consul lock polls where the first
          argument is the attempt number (starting at 1)
          :return: information about the lock if acquired, else `None` if not acquired and not blocking
          :raises InvalidKeyError: raised if the given key is not valid
          :raises LockAcquireTimeoutError: raised if times out waiting for the lock
          """
          ConsulLockManager.validate_key(key)
  
          logger.debug("Creating Consul session...")
          session_id = self.consul_client.session.create(
              lock_delay=0, ttl=self.session_ttl_in_seconds, behavior="delete")
          self._acquiring_session_ids.add(session_id)
          logger.info(f"Created session with ID: {session_id}")
          start_time = monotonic()
  
          @timeout_decorator.timeout(timeout, timeout_exception=LockAcquireTimeoutError)
          def _acquire() -> Optional[ConsulLockInformation]:
              i = 1
              while True:
                  logger.debug("Going to acquire lock")
                  seconds_to_lock = monotonic() - start_time
                  on_before_lock(key)
                  lock_information = self._acquire_lock(key, session_id, seconds_to_lock, metadata)
                  if lock_information is not None:
                      logger.debug("Acquired lock!")
                      return lock_information
                  else:
                      on_lock_already_locked(key)
                      if not blocking:
                          logger.debug("Could not acquire lock (already locked) and not blocking")
                          return None
                      else:
                          logger.debug("Could not acquire lock (already locked)")
                  interval = lock_poll_interval_generator(i)
                  logger.debug(f"Sleeping for {interval}s")
                  sleep(interval)
                  i += 1
  
          lock_information = _acquire()
          self._acquiring_session_ids.remove(session_id)
  
          if lock_information is None:
              self.consul_client.session.destroy(session_id=session_id)
              logger.info(f"Destroyed session (did not acquire the lock)")
  
          return lock_information