97 lines
3.4 KiB
Python
97 lines
3.4 KiB
Python
"""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
|