feat(API): add delegates implementation to Safe API client

ape_safe/client/__init__.py

@@ -1,13 +1,15 @@
 from datetime import datetime
 from functools import reduce
-from typing import Dict, Iterator, Optional, Union, cast
+from typing import Dict, Iterator, List, Optional, Union, cast
 
+from ape.api import AccountAPI
 from ape.types import AddressType, HexBytes, MessageSignature
 from eip712.common import SafeTxV1, SafeTxV2
 
 from ape_safe.client.base import BaseSafeClient
 from ape_safe.client.mock import MockSafeClient
 from ape_safe.client.types import (
+    DelegateInfo,
     ExecutedTxData,
     OperationType,
     SafeApiTxData,
@@ -176,6 +178,54 @@ def estimate_gas_cost(
         gas = result.get("safeTxGas")
         return int(HexBytes(gas).hex(), 16)
 
+    def get_delegates(self) -> Dict[AddressType, List[AddressType]]:
+        url = "delegates"
+        delegates: Dict[AddressType, List[AddressType]] = {}
+
+        while url:
+            response = self._get(url, params={"safe": self.address})
+            data = response.json()
+
+            for delegate_info in map(DelegateInfo.model_validate, data.get("results", [])):
+                if delegate_info.delegator not in delegates:
+                    delegates[delegate_info.delegator] = [delegate_info.delegate]
+                else:
+                    delegates[delegate_info.delegator].append(delegate_info.delegate)
+
+            url = data.get("next")
+
+        return delegates
+
+    def add_delegate(self, delegate: AddressType, label: str, delegator: AccountAPI):
+        msg = self.create_delegate_message(delegate)
+        if not (sig := delegator.sign_message(msg)):
+            raise
+
+        payload = {
+            "safe": self.address,
+            "delegate": delegate,
+            "delegator": delegator.address,
+            "label": label,
+            "signature": sig.encode_rsv().hex(),
+        }
+        response = self._post("delegates", json=payload)
+
+        if not response.ok:
+            raise
+
+    def remove_delegate(self, delegate: AddressType, delegator: AccountAPI):
+        msg = self.create_delegate_message(delegate)
+        if not (sig := delegator.sign_message(msg)):
+            raise
+
+        payload = {
+            "signature": sig.encode_rsv().hex(),
+        }
+        response = self._delete(f"delegates/{delegate}", json=payload)
+
+        if not response.ok:
+            raise
+
 
 __all__ = [
     "ExecutedTxData",

ape_safe/client/base.py

@@ -143,12 +143,15 @@ def session(self) -> requests.Session:
         session.mount("https://", adapter)
         return session
 
-    def _get(self, url: str) -> Response:
-        return self._request("GET", url)
+    def _get(self, url: str, params: Optional[Dict] = None) -> Response:
+        return self._request("GET", url, params=params)
 
     def _post(self, url: str, json: Optional[Dict] = None, **kwargs) -> Response:
         return self._request("POST", url, json=json, **kwargs)
 
+    def _delete(self, url: str, json: Optional[Dict] = None, **kwargs) -> Response:
+        return self._request("DELETE", url, json=json, **kwargs)
+
     @cached_property
     def _http(self):
         return urllib3.PoolManager(ca_certs=certifi.where())

ape_safe/client/types.py

@@ -124,3 +124,10 @@ class ExecutedTxData(UnexecutedTxData):
 
 
 SafeApiTxData = Union[ExecutedTxData, UnexecutedTxData]
+
+
+class DelegateInfo(BaseModel):
+    safe: AddressType
+    delegate: AddressType
+    delegator: AddressType
+    label: str = ""