How to use the spectacles.utils.compose_url function in spectacles

To help you get started, we’ve selected a few spectacles examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github spectacles-ci / spectacles / tests / test_utils.py View on Github external
def test_compose_url_one_path_component():
    url = utils.compose_url(TEST_BASE_URL, ["api"])
    assert url == "https://test.looker.com/api"
github spectacles-ci / spectacles / tests / test_utils.py View on Github external
def test_compose_url_with_extra_slashes():
    url = utils.compose_url(TEST_BASE_URL + "/", ["/api//", "3.0/login/"])
    assert url == "https://test.looker.com/api/3.0/login"
github spectacles-ci / spectacles / spectacles / client.py View on Github external
def create_branch(self, project: str, branch: str, ref: str = "origin/master"):
        """Creates a branch in the given project.

        Args:
            project: Name of the Looker project to use.
            branch: Name of the branch to create.
            ref: The ref to create the branch from.
        """
        logger.debug(
            f"Creating branch '{branch}' on project '{project}' with ref '{ref}'"
        )

        body = {"name": branch, "ref": ref}
        url = utils.compose_url(self.api_url, path=["projects", project, "git_branch"])
        response = self.post(url=url, json=body, timeout=TIMEOUT_SEC)

        try:
            response.raise_for_status()
        except requests.exceptions.HTTPError:
            raise LookerApiError(
                name="unable-to-create-branch",
                title="Couldn't create new Git branch.",
                status=response.status_code,
                detail=(
                    f"Unable to create branch '{branch}' "
                    f"in project '{project}' using ref '{ref}'. "
                    "Confirm the branch doesn't already exist and try again."
                ),
                response=response,
            )
github spectacles-ci / spectacles / spectacles / client.py View on Github external
def get_looker_release_version(self) -> str:
        """Gets the version number of connected Looker instance.

        Returns:
            str: Looker instance release version number (e.g. 6.22.12)

        """
        logger.debug("Checking Looker instance release version")

        url = utils.compose_url(self.api_url, path=["versions"])

        response = self.get(url=url, timeout=TIMEOUT_SEC)
        try:
            response.raise_for_status()
        except requests.exceptions.HTTPError:
            raise LookerApiError(
                name="unable-to-get-version",
                title="Couldn't get Looker's release version.",
                status=response.status_code,
                detail=(
                    "Unable to get the release version of your Looker instance. "
                    "Please try again."
                ),
                response=response,
            )
github spectacles-ci / spectacles / spectacles / client.py View on Github external
def authenticate(self) -> None:
        """Logs in to Looker's API using a client ID/secret pair and an API version.

        Args:
            client_id: Looker API client ID.
            client_secret: Looker API client secret.
            api_version: Desired API version to use for requests.

        """
        logger.debug("Authenticating Looker API credentials")

        url = utils.compose_url(self.api_url, path=["login"])
        body = {"client_id": self.client_id, "client_secret": self.client_secret}
        # This should not use `self.post` or it will create a recursive loop
        response = self.session.post(url=url, data=body, timeout=TIMEOUT_SEC)
        try:
            response.raise_for_status()
        except requests.exceptions.HTTPError:
            raise LookerApiError(
                name="unable-to-authenticate",
                title="Couldn't authenticate to the Looker API.",
                status=response.status_code,
                detail=(
                    f"Unable to authenticate with client ID '{self.client_id}'. "
                    "Check that your credentials are correct and try again."
                ),
                response=response,
            )
github spectacles-ci / spectacles / spectacles / client.py View on Github external
If a ClientError or TimeoutError is received, attempts to retry.

        Args:
            session: Existing asychronous HTTP session.
            query_id: ID of a previously created query to run.

        Returns:
            str: ID for the query task, used to check on the status of the query, which
                is being run asynchronously.

        """
        # Using old-style string formatting so that strings are formatted lazily
        logger.debug("Starting query %d", query_id)
        body = {"query_id": query_id, "result_format": "json_detail"}
        url = utils.compose_url(self.api_url, path=["query_tasks"])

        response = self.post(
            url=url, json=body, params={"cache": "false"}, timeout=TIMEOUT_SEC
        )

        try:
            response.raise_for_status()
        except requests.exceptions.HTTPError:
            raise LookerApiError(
                name="unable-to-launch-query",
                title="Couldn't launch query.",
                status=response.status_code,
                detail=(
                    "Failed to create query task for "
                    f"query '{query_id}'. Please try again."
                ),
github spectacles-ci / spectacles / spectacles / client.py View on Github external
def get_active_branch(self, project: str) -> JsonDict:
        """Gets the active branch for the user in the given project.

        Args:
            project: Name of the Looker project to use.

        Returns:
            str: Name of the active branch
        """
        logger.debug(f"Getting active branch for project '{project}'")
        url = utils.compose_url(self.api_url, path=["projects", project, "git_branch"])
        response = self.get(url=url, timeout=TIMEOUT_SEC)

        try:
            response.raise_for_status()
        except requests.exceptions.HTTPError:
            raise LookerApiError(
                name="unable-to-get-active-branch",
                title="Couldn't determine active Git branch.",
                status=response.status_code,
                detail=(
                    f"Unable to get active branch for project '{project}'. "
                    "Please check that the project exists and try again."
                ),
                response=response,
            )
github spectacles-ci / spectacles / spectacles / client.py View on Github external
def get_all_branches(self, project: str) -> List[str]:
        """Returns a list of git branches in the project repository.

        Args:
            project: Name of the Looker project to use.
        """
        logger.debug(f"Getting all Git branches in project '{project}'")
        url = utils.compose_url(
            self.api_url, path=["projects", project, "git_branches"]
        )
        response = self.get(url=url, timeout=TIMEOUT_SEC)
        try:
            response.raise_for_status()
        except requests.exceptions.HTTPError:
            raise LookerApiError(
                name="unable-to-get-branches",
                title="Couldn't get all Git branches.",
                status=response.status_code,
                detail=(
                    f"Unable to get all Git branches in project '{project}'. "
                    "Please try again."
                ),
                response=response,
            )
github spectacles-ci / spectacles / spectacles / client.py View on Github external
def cancel_query_task(self, query_task_id: str):
        """Cancels a query task.

        Args:
            query_task_id: ID for the query task to cancel.

        """
        logger.debug(f"Cancelling query task: {query_task_id}")
        url = utils.compose_url(self.api_url, path=["running_queries", query_task_id])
        self.delete(url=url, timeout=TIMEOUT_SEC)