skilljar_client.py

import logging
import httpx
from urllib.parse import urlunsplit, urlencode

logger = logging.getLogger(__name__)


class SkilljarAdminError(Exception):
    pass


class SkilljarAdmin:
    def __init__(self, domain: str, key: str):
        self._api_url = "https://api.skilljar.com/v1"
        self._domain = domain
        self._key = key

        self.start_session()

    @property
    def api_url(self) -> str:
        return self._api_url

    @property
    def domain(self) -> str:
        return self._domain

    def start_session(self):
        self.s = httpx.AsyncClient()
        self.s.auth = (self._key, "")

    async def post(self, path: str, json: dict) -> dict:
        req = await self.s.post(f"{self.api_url}{path}", json=json)

        try:
            req.raise_for_status()
        except httpx.HTTPError as e:
            raise SkilljarAdminError({e})

        return req.json()

    async def put(self, path: str, json: dict) -> dict:
        req = await self.s.put(f"{self.api_url}{path}", json=json)

        try:
            req.raise_for_status()
        except httpx.HTTPError as e:
            logger.error("SkilljarAdmin Error: " + str(e))
            raise SkilljarAdminError({e})

        return req.json()

    # Set the class user by creating one. If user is already created, the API will return its info
    async def create_user(self, payload: dict) -> tuple:
        """
        Sample payload: {
                "email": "john.doe@myacme.com",
                "first_name": "John",
                "last_name": "Doe"
            }
        """
        user_info_payload = {"user": payload}
        user_info = await self.post(
            path=f"/domains/{self.domain}/users", json=user_info_payload
        )

        try:
            return user_info["user"]["id"], user_info["login_token"]
        except KeyError as e:
            logger.error("Unable to load User ID/Login Token from response: " + str(e))
            raise SkilljarAdminError({e})

    async def update_user(self, user_id: str, payload: dict) -> dict:
        """
        Sample payload: {
                "email": "john.doe@myacme.com",
                "first_name": "John",
                "last_name": "Doe"
            }
        """
        return await self.put(path=f"/users/{user_id}", json=payload)

    async def get_group_id(self, group: str) -> str:
        req = await self.post(path="/groups", json={"name": group})

        try:
            group_id = req["id"]
        except KeyError as e:
            logger.error("Unable to load Group ID from response: " + str(e))
            raise SkilljarAdminError({e})

        return group_id

    async def add_group_member(self, group_name: str, user_id: str) -> dict:
        known_group_ids = {
            "name": "group_id",
        }
        # FIXME
        group_id = known_group_ids.get(group_name, await self.get_group_id(group_name))

        json = {"user": {"id": user_id}}
        return await self.post(path=f"/groups/{group_id}/users", json=json)

    def user_signin_uri(self, token: str) -> str:
        params = {"token": token}
        signin_uri = urlunsplit(
            ("https", self.domain, "/auth/login/callback", urlencode(params), "")
        )

        return signin_uri