Refactor device blocking logic in DeviceBlockAPIView to utilize Omada BlockDeviceResponse for blocking in omada first
Some checks failed
Build and Push Docker Images / Build and Push Docker Images (push) Failing after 3m42s

This commit is contained in:
2025-06-23 23:41:59 +05:00
parent 570cf80019
commit 39da124214
2 changed files with 39 additions and 11 deletions

View File

@ -1,5 +1,12 @@
import requests
from apibase.env import env
from dataclasses import dataclass
@dataclass
class BlockDeviceResponse:
errorCode: int
msg: str
class Omada:
@ -92,7 +99,9 @@ class Omada:
except requests.RequestException as e:
print(f"Error adding devices to Omada: {e}")
def block_device(self, mac_address: str, operation: str = "block"):
def block_device(
self, mac_address: str, operation: str = "block"
) -> BlockDeviceResponse:
"""
Block a device in Omada.
:param mac_address: The MAC address of the device to block.
@ -103,10 +112,23 @@ class Omada:
url,
headers={"X-API-Key": str(self.api_key)},
)
if response.status_code == 200:
data = response.json()
return data
print("Response Data: ", data)
if data["errorCode"] == 0:
return BlockDeviceResponse(
errorCode=data["errorCode"],
msg=data["msg"]
if "msg" in data
else "Device blocked successfully.",
)
else:
print(f"Failed to block device: {response.text}")
return BlockDeviceResponse(
errorCode=data["errorCode"],
msg=data["msg"] if "msg" in data else "Failed to block device.",
)
except requests.RequestException as e:
print(f"Error blocking device in Omada: {e}")
return BlockDeviceResponse(
errorCode=-1, msg=f"Error blocking device in Omada: {e}"
)

View File

@ -127,16 +127,22 @@ class DeviceBlockAPIView(StaffEditorPermissionMixin, generics.UpdateAPIView):
return Response({"message": "Blocked field is required."}, status=400)
if not isinstance(blocked, bool):
return Response({"message": "Blocked field must be a boolean."}, status=400)
instance.blocked = blocked
omada_client = Omada()
omada_client.block_device(
blocked = omada_client.block_device(
instance.mac, operation="block" if blocked else "unblock"
)
if blocked.errorCode == 0:
instance.blocked = blocked
instance.save()
serializer = self.get_serializer(instance, data=request.data, partial=False)
serializer.is_valid(raise_exception=True)
self.perform_update(serializer)
return Response(serializer.data)
else:
return Response(
{"message": blocked.msg},
status=status.HTTP_400_BAD_REQUEST,
)
class DeviceDestroyAPIView(StaffEditorPermissionMixin, generics.DestroyAPIView):