Skip to content

utils

Utility functions for interacting with dapla-team-api.

delete_resource(request)

Get a given resource (Team/Group/User/a list) from the API.

Source code in dapla_team_cli/api/utils.py
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def delete_resource(request: DaplaTeamApiRequest) -> Result[Dict[str, Any], str]:
    """Get a given resource (Team/Group/User/a list) from the API."""
    token = get_token()
    logger.debug(f"Endpoint is {request.endpoint}")
    if request.body is not None:
        json_body = request.body.json()
        logger.debug(f"Body is {request.body.json()}")
    else:
        json_body = None

    try:
        response = requests.delete(
            url=request.endpoint,
            headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
            data=json_body,
            timeout=300,
        )
    except requests.RequestException as e:
        logger.debug("exception thrown in get_resource accessing %s, exception: %s", request.endpoint, str(e))
        return Failure("Something went wrong trying to access the API")

    pretty_reponse = json.dumps(response.json(), indent=4)
    logger.debug(f"Response: \n {pretty_reponse}")

    if response.status_code not in (200, 201):
        logger.debug(
            "delete_resource got status code %s accessing %s | Reason: %s | Response body: %s",
            response.status_code,
            request.endpoint,
            response.reason,
            response.content.decode("utf-8"),
        )
        return Failure(f"Error trying to access {request.endpoint}: {response.status_code} {response.reason}")

    return Success(response.json())

get_resource(request)

Get a given resource (Team/Group/User/a list) from the API.

Source code in dapla_team_cli/api/utils.py
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def get_resource(request: DaplaTeamApiRequest) -> Result[Dict[str, Any], str]:
    """Get a given resource (Team/Group/User/a list) from the API."""
    token = get_token()
    logger.debug(f"Endpoint is {request.endpoint}")
    try:
        response = requests.get(
            request.endpoint,
            headers={
                "Authorization": f"Bearer {token}",
            },
            timeout=10,
        )
    except requests.RequestException as e:
        logger.debug("exception thrown in get_resource accessing %s, exception: %s", request.endpoint, str(e))
        return Failure("Something went wrong trying to access the API")

    if response.status_code != 200:
        logger.debug(
            "get_resource got status code %s accessing %s | Reason: %s | Response body: %s",
            response.status_code,
            request.endpoint,
            response.reason,
            response.content.decode("utf-8"),
        )
        return Failure(f"Error trying to access {request.endpoint}: {response.status_code} {response.reason}")

    return Success(response.json())

post_resource(request)

Get a given resource (Team/Group/User/a list) from the API.

Source code in dapla_team_cli/api/utils.py
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def post_resource(request: DaplaTeamApiRequest) -> Result[Dict[str, Any], str]:
    """Get a given resource (Team/Group/User/a list) from the API."""
    logger.debug(f"Endpoint is {request.endpoint}")
    if request.body is not None:
        json_body = request.body.json()
        logger.debug(f"Body is {json_body}")
    else:
        json_body = None

    token = get_token()
    try:
        response = requests.post(
            url=request.endpoint,
            headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
            data=json_body,
            timeout=300,
        )
    except requests.RequestException as e:
        logger.debug("exception thrown in get_resource accessing %s, exception: %s", request.endpoint, str(e))
        return Failure("Something went wrong trying to access the API")

    pretty_reponse = json.dumps(response.json(), indent=4)
    logger.debug(f"Response: \n {pretty_reponse}")

    if response.status_code not in (200, 201):
        logger.debug(
            "post_resource got status code %s accessing %s | Reason: %s | Response body: %s",
            response.status_code,
            request.endpoint,
            response.reason,
            response.content.decode("utf-8"),
        )
        return Failure(f"Error trying to access {request.endpoint}: {response.status_code} {response.reason}")

    return Success(response.json())