How to use the httpx.post function in httpx

To help you get started, weā€™ve selected a few httpx 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 strawberry-graphql / strawberry / .github / release-check-action / check.py View on Github external
except InvalidReleaseFileError:
        exit_code = 2
        status = "INVALID"

mutation = """mutation AddReleaseComment($input: AddReleaseFileCommentInput!) {
  addReleaseFileComment(input: $input)
}"""

mutation_input = {
    "prNumber": event_data["number"],
    "status": status,
    "releaseInfo": release_info,
}


response = httpx.post(
    API_URL,
    json={"query": mutation, "variables": {"input": mutation_input}},
    timeout=120,
)
response.raise_for_status()

print(f"Status is {status}")
sys.exit(exit_code)
github compute-tooling / compute-studio / workers / cs_workers / models / executors / job.py View on Github external
def sim_handler(task_id, meta_param_dict, adjustment):
    from cs_config import functions

    outputs = functions.run_model(meta_param_dict, adjustment)
    print("got result")
    outputs = cs_storage.serialize_to_json(outputs)
    print("storing results")
    for i in range(3):
        try:
            resp = httpx.post(
                "http://outputs-processor/write/",
                json={"task_id": task_id, "outputs": outputs},
                timeout=120.0,
            )
            break
        except Exception as e:
            print(i, e)

    print("got resp", resp.status_code, resp.url)
    assert resp.status_code == 200, f"Got code: {resp.status_code}"
    return resp.json()
github msys2 / msys2-web / app / main.py View on Github external
def trigger_appveyor_build(account: str, project: str, token: str) -> str:
    """Returns an URL for the build or raises RequestException"""

    r = httpx.post(
        "https://ci.appveyor.com/api/builds",
        json={
            "accountName": account,
            "projectSlug": project,
            "branch": "master",
        },
        headers={
            "Authorization": "Bearer " + token,
        },
        timeout=REQUEST_TIMEOUT)
    r.raise_for_status()

    try:
        build_id = r.json()['buildId']
    except (ValueError, KeyError):
        build_id = 0
github strawberry-graphql / strawberry / .github / release-check-action / github.py View on Github external
def add_or_edit_comment(github_event_data: dict, comment: str):
    current_comments = get_comments(github_event_data)

    previous_comment = next(
        (comment for comment in current_comments if is_release_check_comment(comment)),
        None,
    )

    method = httpx.patch if previous_comment else httpx.post
    url = (
        previous_comment["url"]
        if previous_comment
        else get_comments_link(github_event_data)
    )

    request = method(
        url,
        headers={"Authorization": f"token {GITHUB_TOKEN}"},
        json={"body": comment + SIGNATURE},
    )

    if request.status_code >= 400:
        print(request.text)
        print(request.status_code)
github compute-tooling / compute-studio / workers / cs_workers / models / manage.py View on Github external
def stage_app(self, app):
        resp = httpx.post(
            f"{self.config.cs_url}/publish/api/{app['owner']}/{app['title']}/deployments/",
            json={"staging_tag": self.staging_tag},
            headers={"Authorization": f"Token {self.cs_api_token}"},
        )
        assert (
            resp.status_code == 200
        ), f"Got: {resp.url} {resp.status_code} {resp.text}"

        sys.stdout.write(resp.json()["staging_tag"])
github strawberry-graphql / strawberry / .github / release-check-action / github.py View on Github external
current_labels_url_by_name = {
        label["name"]: label["url"]
        for label in github_event_data["pull_request"]["labels"]
    }

    current_labels = set(current_labels_url_by_name.keys())

    release_labels_to_remove = [
        label
        for label in current_labels
        if label.startswith("bot:release-type-") and label != new_release_label
    ]
    labels_to_remove.update(release_labels_to_remove)

    if not current_labels.issuperset(labels_to_add):
        request = httpx.post(
            labels_url,
            headers={"Authorization": f"token {GITHUB_TOKEN}"},
            json={"labels": list(labels_to_add)},
        )

        if request.status_code >= 400:
            print(request.text)
            print(request.status_code)

    if current_labels.issuperset(labels_to_remove):
        for label in labels_to_remove:
            request = httpx.delete(
                current_labels_url_by_name[label],
                headers={"Authorization": f"token {GITHUB_TOKEN}"},
            )
github compute-tooling / compute-studio / workers / cs_workers / models / manage.py View on Github external
def promote_app(self, app):
        resp = httpx.get(
            f"{self.config.cs_url}/publish/api/{app['owner']}/{app['title']}/deployments/",
            headers={"Authorization": f"Token {self.cs_api_token}"},
        )
        assert (
            resp.status_code == 200
        ), f"Got: {resp.url} {resp.status_code} {resp.text}"
        staging_tag = resp.json()["staging_tag"]
        resp = httpx.post(
            f"{self.config.cs_url}/publish/api/{app['owner']}/{app['title']}/deployments/",
            json={"latest_tag": staging_tag or self.tag, "staging_tag": None},
            headers={"Authorization": f"Token {self.cs_api_token}"},
        )
        assert (
            resp.status_code == 200
        ), f"Got: {resp.url} {resp.status_code} {resp.text}"

        sys.stdout.write(resp.json()["latest_tag"])