def get_service_perimeter_status()

in dialogflow-cx/vpc-sc-demo/backend/status_utilities.py [0:0]

• 36 lines of code
• 6 McCabe index (conditional complexity)

def get_service_perimeter_status(token, project_id, access_policy_name):
    """Get service perimeter status using accesscontextmanager API."""
    headers = {}
    headers["x-goog-user-project"] = project_id
    headers["Authorization"] = f"Bearer {token}"
    response = get_service_perimeter_data_uri(token, project_id, access_policy_name)
    if "response" in response:
        return response
    service_perimeter_data_uri = response["uri"]
    result = requests.get(service_perimeter_data_uri, headers=headers, timeout=10)
    if result.status_code != 200:
        logger.info("  accesscontextmanager API rejected request: %s", result.text)
        if (result.json()["error"]["status"] == "PERMISSION_DENIED") and (
            result.json()["error"]["message"].startswith(
                "Access Context Manager API has not been used in project"
            )
        ):
            # Potential bug: should return a dict?
            return flask.Response(
                status=200,
                response=json.dumps(
                    {
                        "status": "BLOCKED",
                        "reason": "ACCESS_CONTEXT_MANAGER_API_DISABLED",
                    }
                ),
            )
        if result.json()["error"]["status"] == "PERMISSION_DENIED":
            response = flask.Response(
                status=200,
                response=json.dumps(
                    {"status": "BLOCKED", "reason": "PERMISSION_DENIED"}
                ),
            )
            return {"response": response}
        response = flask.Response(status=result.status_code, response=result.text)
        return {"response": response}
    return result.json()