Initial commit

polaris_client.py

@@ -0,0 +1,96 @@
+"""Minimal client for the Polaris Leap (PAPI) Notices endpoint."""
+
+import json
+
+import requests
+
+
+class PolarisAuthError(Exception):
+    pass
+
+
+class PolarisRequestError(Exception):
+    pass
+
+
+class PolarisClient:
+    def __init__(
+        self,
+        server,
+        site_domain,
+        organization_id,
+        workstation_id,
+        language="eng",
+        product_id=7,
+    ):
+        self.auth_url = f"https://{server}/Polaris.ApplicationServices/api/v1/{language}/{product_id}/authentication/staffuser"
+        self.base_url = f"https://{server}/Polaris.ApplicationServices/api/v1/{language}/{product_id}/{site_domain}/{organization_id}/{workstation_id}"
+        self.site_domain = site_domain
+        self.access_token = None
+        self.access_secret = None
+
+    def authenticate(self, username, password):
+        """Exchange a staff username/password for an AccessToken/AccessSecret.
+
+        Username must include the domain, e.g. MYLIB\\Aladdin.
+
+        Note: per the live API help page (view-source, not the rendered page --
+        it's a tooltip on the "..." in the route), the staff-authentication route
+        only includes {version}/{language}/{productId} -- no siteDomain,
+        organizationId, or workstationId, since those describe a *logged-on*
+        session and you're not logged on yet at this point.
+        """
+        response = requests.post(
+            self.auth_url,
+            auth=(username, password),
+            headers={"Accept": "application/json"},
+        )
+        if response.status_code == 401:
+            raise PolarisAuthError("Authentication failed: check username/password.")
+        if not response.ok:
+            raise PolarisAuthError(
+                f"Authentication failed ({response.status_code}): {response.text}"
+            )
+
+        data = response.json()
+        self.access_token = data["AccessToken"]
+        self.access_secret = data["AccessSecret"]
+        return data
+
+    def _auth_header(self):
+        if not self.access_token or not self.access_secret:
+            raise PolarisAuthError("Not authenticated yet -- call authenticate() first.")
+        return f"PAS {self.site_domain}:{self.access_token}:{self.access_secret}"
+
+    def post_notices(self, notice_type, organization_ids, delivery_option=1):
+        """POST notices for the given notice type to the given org IDs.
+
+        organization_ids: an iterable/list of integer org IDs.
+        delivery_option: only 1 (Mail) is currently supported by the API.
+
+        Returns True/False as reported by Polaris.
+        """
+        org_ids_csv = ",".join(str(oid) for oid in organization_ids)
+
+        response = requests.post(
+            f"{self.base_url}/notices/post",
+            params={"noticeType": notice_type, "deliveryOption": delivery_option},
+            headers={
+                "Authorization": self._auth_header(),
+                "Content-Type": "application/json",
+                "Accept": "application/json",
+            },
+            data=json.dumps(org_ids_csv),
+        )
+        if not response.ok:
+            raise PolarisRequestError(
+                f"Notice post failed ({response.status_code}): {response.text}"
+            )
+
+        result = response.json()
+        if result is not True:
+            raise PolarisRequestError(
+                f"Polaris reported failure (returned {result!r}) for noticeType={notice_type}, "
+                f"organizationIds={org_ids_csv}"
+            )
+        return result