Assign Sensor to a New Group

from typing import Any

import requests  
from oauthlib.oauth2 import BackendApplicationClient  
from requests.auth import HTTPBasicAuth  
from requests_oauthlib import OAuth2Session

# Global constants

AUTH_TOKEN_URL = "<https://sso.common.cloud.hpe.com/as/token.oauth2">  
UXI_API_BASE_URL = "<https://api.capenetworks.com/networking-uxi/v1alpha1">

# Customer specific constants

CLIENT_ID = "\<CLIENT_ID>"  
CLIENT_SECRET = "\<CLIENT_SECRET>"

# Example specific constants

# The ID of the parent group to use when creating the new group. The new group will be created as a child of this group.

# This can be retrieved from the API: /networking-uxi/v1alpha1/groups.

PARENT_GROUP_ID = "\<PARENT_GROUP_ID>" # eg: 9bfded8bd7cb

# The ID of the sensor to assign to the new group.

# This can be retrieved from the API: /networking-uxi/v1alpha1/sensors.

SENSOR_ID = "\<SENSOR_ID>" # eg: 9108f997-32a0-45ca-ab00-c8c93273c332

def main():  
    api_token = get_api_token()

```
child_group_id = create_group(
    api_token=api_token,
    group_name="Child Group",
    parent_id=PARENT_GROUP_ID,
)
if not child_group_id:
    exit(1)

all_groups = get_groups(api_token=api_token)

# Ensure that at least 1 network is assigned to the new group.
networks_assigned_to_group = get_networks_assigned_to_group(
    api_token=api_token, all_groups=all_groups, group_id=child_group_id
)
if not networks_assigned_to_group:
    wireless_networks = get_wireless_networks(api_token=api_token)
    if not wireless_networks:
        return

    # assign a wireless network to the new group
    assign_network_to_group(
        api_token=api_token,
        group_id=child_group_id,
        network_id=wireless_networks[0]["id"],
    )

assign_sensor_to_group(
    api_token=api_token, sensor_id=SENSOR_ID, group_id=child_group_id
)
```
def get_api_token() -> str:  
    client = BackendApplicationClient(CLIENT_ID)  
    oauth = OAuth2Session(client=client)  
    auth = HTTPBasicAuth(CLIENT_ID, CLIENT_SECRET)  
    token = oauth.fetch_token(token_url=AUTH_TOKEN_URL, auth=auth)

```
return token["access_token"]
```
def create_group(  
    api_token: str, group_name: str, parent_id: str  
) -> str | None:  
    HEADERS = {  
        "Authorization": f"Bearer {api_token}",  
        "Content-Type": "application/json",  
    }  
    request_body = {"name": group_name, "parentId": parent_id}  
    try:  
        response = requests.post(f"{UXI_API_BASE_URL}/groups", headers=HEADERS, json=request_body)  
        response.raise_for_status()  
        response_body = response.json()  
        print(f"Successfully created group: {response_body['name']}")

```
    return response_body["id"]
except requests.exceptions.HTTPError as e:
    print(f"Failed to create group: {e.response.text}")
```
def assign_network_to_group(  
    api_token: str, group_id: str, network_id: str  
) -> str | None:  
    HEADERS = {  
        "Authorization": f"Bearer {api_token}",  
        "Content-Type": "application/json",  
    }  
    request_body = {"groupId": group_id, "networkId": network_id}

```
try:
    response = requests.post(
        f"{UXI_API_BASE_URL}/network-group-assignments", headers=HEADERS, json=request_body
    )
    response.raise_for_status()
    response_body = response.json()
    print(f"Successfully assigned network '{network_id}' to group '{group_id}'")

    return response_body["id"]
except requests.exceptions.HTTPError as e:
    print(f"Failed to assign network '{network_id}' to group '{group_id}': {e.response.text}")
```
def assign_sensor_to_group(api_token: str, sensor_id: str, group_id: str) -> str | None:  
    HEADERS = {  
        "Authorization": f"Bearer {api_token}",  
        "Content-Type": "application/json",  
    }

```
payload = {"sensorId": sensor_id, "groupId": group_id}
try:
    response = requests.post(
        f"{UXI_API_BASE_URL}/sensor-group-assignments",
        headers=HEADERS,
        json=payload,
    )
    response_body = response.json()
    response.raise_for_status()
    print(f"Successfully assigned sensor '{sensor_id}' to group '{group_id}'.")

    return response_body["id"]
except requests.exceptions.HTTPError as e:
    print(f"Failed to assign sensor '{sensor_id}' to group '{group_id}': {e.response.text}")
```
def get_networks_assigned_to_group(  
    api_token: str, all_groups: list\[dict[str, Any]], group_id: str  
) -> list\[dict[str, Any]]:  
    # Groups inherit the network assignments from their parents.  
    # To know which networks are assigned to a particular group, we need to know  
    # which networks are directly assigned to the group itself as well as any of the  
    # groups parents.  
    group_parents = get_group_parents(all_groups=all_groups, group_id=group_id)  
    network_group_assignments = get_network_group_assignments(api_token=api_token)

```
group_lineage = [parent["id"] for parent in group_parents] + [group_id]

return [
    assignment
    for assignment in network_group_assignments
    if assignment["group"]["id"] in group_lineage
]
```
def get_group_parents(all_groups: list\[dict[str, Any]], group_id: str) -> list\[dict[str, Any]]:  
    group_map = {group["id"]\: group for group in all_groups}

```
def _get_parent(group_id: str) -> list[dict[str, Any]]:
    group = group_map.get(group_id)
    if not group or not group.get("parent", None):
        return []

    parent_id = group["parent"]["id"]
    return _get_parent(parent_id) + [group_map[parent_id]]

return _get_parent(group_id)
```
def get_groups(api_token: str) -> list\[dict[str, Any]]:  
    return \_get_items(path="groups", api_token=api_token)

def get_network_group_assignments(api_token: str) -> list\[dict[str, Any]]:  
    return \_get_items(path="network-group-assignments", api_token=api_token)

def get_wireless_networks(api_token: str) -> list\[dict[str, Any]]:  
    return \_get_items(path="wireless-networks", api_token=api_token)

def \_get_items(path: str, api_token: str) -> list\[dict[str, Any]]:  
    HEADERS = {  
        "Authorization": f"Bearer {api_token}",  
    }  
    items: list\[dict[str, Any]] = \[]  
    next_cursor = None

```
# This while loop exists to keep making requests until all data for the specified resource is fetched.
while True:
    try:
        params = {"next": next_cursor} if next_cursor else {}
        response = requests.get(
            f"{UXI_API_BASE_URL}/{path}", headers=HEADERS, params=params
        )
        response.raise_for_status()
        response_body: dict[str, Any] = response.json()
        response_items: list[dict[str, Any]] = response_body.get("items", [])
        next_cursor = response_body.get("next")

        items.extend(response_items)

        if not next_cursor or not response_items:
            break
    except requests.exceptions.HTTPError as e:
        print(f"Error fetching items ({path}): {e}")
        break

return items
```
if **name** == "**main**":  
    main()