File size: 2,508 Bytes
027d6cc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 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 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 | """
Manage Gated Access Requests for bbkdevops/Fiber-MoE-Symplectic-Gating-Research
Uses official huggingface_hub Gated Model API and HTTP Endpoints.
"""
from huggingface_hub import HfApi, get_token
import requests
import json
REPO_ID = "bbkdevops/Fiber-MoE-Symplectic-Gating-Research"
BASE_URL = f"https://huggingface.co/api/models/{REPO_ID}/user-access-request"
class GatedModelController:
def __init__(self, repo_id: str = REPO_ID):
self.repo_id = repo_id
self.token = get_token()
self.api = HfApi(token=self.token)
def _headers(self):
return {"Authorization": f"Bearer {self.token}"}
def list_pending(self):
"""Retrieve list of pending access requests."""
url = f"{BASE_URL}/pending"
res = requests.get(url, headers=self._headers())
if res.status_code == 200:
return res.json()
return {"status_code": res.status_code, "msg": res.text}
def list_accepted(self):
"""Retrieve list of accepted access requests."""
url = f"{BASE_URL}/accepted"
res = requests.get(url, headers=self._headers())
if res.status_code == 200:
return res.json()
return {"status_code": res.status_code, "msg": res.text}
def grant_user(self, username: str):
"""Directly grant a user access to the model."""
url = f"{BASE_URL}/grant"
payload = {"user": username}
res = requests.post(url, headers=self._headers(), json=payload)
return {"status_code": res.status_code, "response": res.text}
def handle_request(self, username: str, status: str = "accepted", reason: str = ""):
"""Handle request: status in ['accepted', 'rejected', 'pending', 'reset']."""
url = f"{BASE_URL}/handle"
payload = {"status": status, "user": username}
if reason:
if status == "rejected":
payload["rejectionReason"] = reason[:200]
elif status == "reset":
payload["resetReason"] = reason[:200]
res = requests.post(url, headers=self._headers(), json=payload)
return {"status_code": res.status_code, "response": res.text}
if __name__ == "__main__":
controller = GatedModelController()
print("Checking Gated Model Requests for:", REPO_ID)
pending = controller.list_pending()
print("Pending Requests:", json.dumps(pending, indent=2))
accepted = controller.list_accepted()
print("Accepted Requests:", json.dumps(accepted, indent=2))
|