mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
35 lines
1.2 KiB
Python
35 lines
1.2 KiB
Python
import requests
|
|
|
|
from datatypes.magicplan.api.response import MagicPlanPlan, PlanSummary, PlansListResponse
|
|
|
|
_BASE_URL = "https://cloud.magicplan.app/api/v2"
|
|
|
|
|
|
class MagicPlanClient:
|
|
def __init__(self, customer_id: str, api_key: str) -> None:
|
|
self._session = requests.Session()
|
|
self._session.headers.update({"customer": customer_id, "key": api_key})
|
|
|
|
def get_plans(self) -> list[PlanSummary]:
|
|
all_plans: list[PlanSummary] = []
|
|
page = 1
|
|
while True:
|
|
r = self._session.get(f"{_BASE_URL}/workgroups/plans", params={"page": page})
|
|
r.raise_for_status()
|
|
response = PlansListResponse.model_validate(r.json()["data"])
|
|
all_plans.extend(response.plans)
|
|
if not response.paging.next_page:
|
|
break
|
|
page += 1
|
|
return all_plans
|
|
|
|
def get_plan(self, plan_id: str) -> MagicPlanPlan:
|
|
return MagicPlanPlan.model_validate(self._fetch_plan(plan_id).json()["data"])
|
|
|
|
def get_plan_raw(self, plan_id: str) -> bytes:
|
|
return self._fetch_plan(plan_id).content
|
|
|
|
def _fetch_plan(self, plan_id: str) -> requests.Response:
|
|
r = self._session.get(f"{_BASE_URL}/plans/get/{plan_id}")
|
|
r.raise_for_status()
|
|
return r
|