def create_queue()

in common/cloud_task/cloud_task_publisher.py [0:0]

• 15 lines of code
• 1 McCabe index (conditional complexity)

    def create_queue(self) -> Queue:
        """
        Creates a queue with Google Cloud Queues.
        """
        parent = f"projects/{self.project}/locations/{self.location}"
        rate_limits = RateLimits({
            "max_dispatches_per_second": self.max_rps,
        })
        queue = Queue({
            "name": self._queue_fqn,
            "rate_limits": rate_limits
        })
        result = self._cloud_task_client.create_queue(
            request={"parent": parent, "queue": queue}
        )
        time.sleep(self._wait_after_queue_creation)

        self._logger.info(f"Created queue: {self._queue_fqn}")

        return result