client.py

"""Client for connecting to CIM API"""
import logging
import urllib.parse
from typing import Tuple, Union, List, Iterable, Set, Any, Dict, Optional

import requests

from cim_client.models import (
    BaseModel,
    Organisation,
    OrganisationImportResult,
    OrganisationList,
    PersonBase,
    PersonList,
    PersonBase_T,
)

logger = logging.getLogger(__name__)


JsonType = Any


def _serialise_upsert_person(
    person: PersonBase_T,
) -> Dict[str, Any]:
    """
    Make sure empty phone numbers are excluded.

    A phone field is reset when the corresponding key is omitted.
    Neither "" nor null are allowed values.
    """
    empty_phone_fields = {
        f
        for f in {
            "job_mobile",
            "job_phone",
            "secret_number",
            "private_mobile",
            "private_phone",
        }
        if not person.__dict__[f]
    }

    return person.dict(exclude=empty_phone_fields or None)  # type: ignore[arg-type]


def merge_dicts(*dicts: Optional[Dict[Any, Any]]) -> Dict[Any, Any]:
    """
    Combine a series of dicts without mutating any of them.

    >>> merge_dicts({'a': 1}, {'b': 2})
    {'a': 1, 'b': 2}
    >>> merge_dicts({'a': 1}, {'a': 2})
    {'a': 2}
    >>> merge_dicts(None, None, None)
    {}
    """
    combined = dict()
    for d in dicts:
        if not d:
            continue
        for k in d:
            combined[k] = d[k]
    return combined


class CimEndpoints:
    def __init__(
        self,
        url: str,
        upsert_persons_url: Optional[str] = None,
        delete_persons_url: Optional[str] = None,
        import_organisations_url: Optional[str] = None,
    ):
        """ Get endpoints relative to the CIM API URL. """
        self.baseurl = url
        self.upsert_persons_url = upsert_persons_url or "?ws=contacts/upsert/1.0"
        self.delete_persons_url = delete_persons_url or "?ws=contacts/delete/1.0"
        self.import_organisations_url = (
            import_organisations_url or "?ws=contacts/organisations/1.0"
        )

    def __repr__(self) -> str:
        return "{cls.__name__}({url!r})".format(cls=type(self), url=self.baseurl)

    def upsert_persons(self) -> str:
        return urllib.parse.urljoin(self.baseurl, self.upsert_persons_url)

    def delete_persons(self) -> str:
        return urllib.parse.urljoin(self.baseurl, self.delete_persons_url)

    def get_upsert_persons_schema(self) -> str:
        return urllib.parse.urljoin(self.baseurl, self.upsert_persons_url)

    def get_delete_persons_schema(self) -> str:
        return urllib.parse.urljoin(self.baseurl, self.delete_persons_url)

    def import_organisations(self) -> str:
        return urllib.parse.urljoin(self.baseurl, self.import_organisations_url)

    def get_import_organisations_schema(self) -> str:
        return urllib.parse.urljoin(self.baseurl, self.import_organisations_url)


class CimClient(object):
    default_headers = {
        "Accept": "application/json",
    }

    def __init__(
        self,
        endpoints: Union[CimEndpoints, Dict[str, str]],
        headers: Union[None, Dict[str, str]] = None,
        return_objects: bool = True,
        use_sessions: bool = True,
        auth: Optional[Tuple[str, str]] = None,
    ):
        """
        CIM API client.
        :param CimEndpoints endpoints: API endpoints
        :param dict headers: Append extra headers to all requests
        :param bool return_objects: Return objects instead of raw JSON
        :param bool use_sessions: Keep HTTP connections alive (default True)
        :param tuple auth: (username, password) for simple http auth
        """
        if isinstance(endpoints, dict):
            endpoints = CimEndpoints(**endpoints)

        self.endpoints = endpoints

        if isinstance(auth, list):
            auth = tuple(auth)
        self.auth = auth if auth else None

        self.headers = merge_dicts(self.default_headers, headers)
        self.return_objects = return_objects
        if use_sessions:
            self.session = requests.Session()
        else:
            self.session = requests  # type: ignore

    def _build_request_headers(self, headers: Dict[str, str]) -> Dict[str, str]:
        request_headers = {}
        for h in self.headers:
            request_headers[h] = self.headers[h]
        for h in headers or ():
            request_headers[h] = headers[h]
        return request_headers

    def call(
        self,
        method_name: str,
        url: str,
        headers: Optional[Dict[str, str]] = None,
        params: Optional[Dict[str, str]] = None,
        return_response: bool = True,
        **kwargs: Any
    ) -> Union[requests.Response, JsonType]:
        headers = self._build_request_headers(headers or {})
        if params is None:
            params = {}
        logger.debug(
            "Calling %s %s with params=%r",
            method_name,
            urllib.parse.urlparse(url).path,
            params,
        )
        r = self.session.request(
            method_name, url, headers=headers, params=params, **kwargs
        )
        if r.status_code in (500, 400, 401):
            logger.warning("Got HTTP %d: %r", r.status_code, r.content)
        if return_response:
            return r
        r.raise_for_status()
        return r.json()

    def get(self, url: str, **kwargs: Any) -> Union[requests.Response, JsonType]:
        return self.call("GET", url, **kwargs)

    def put(self, url: str, **kwargs: Any) -> Union[requests.Response, JsonType]:
        return self.call("PUT", url, **kwargs)

    def post(self, url: str, **kwargs: Any) -> Union[requests.Response, JsonType]:
        return self.call("POST", url, **kwargs)

    def object_or_data(
        self, cls: BaseModel, data: JsonType
    ) -> Union[BaseModel, JsonType]:
        if not self.return_objects:
            return data
        return cls.from_dict(data)

    def get_upsert_persons_schema(self) -> JsonType:
        """GETs the current schema from CIM

        The schema can change depending on the installation.

        It may be a good idea to setup some sort of daily(?)cron job that
        checks whether the schema has changed lately so you don't update
        using an outdated schema.
        """
        url = self.endpoints.get_upsert_persons_schema()
        response = self.get(
            url,
            auth=self.auth,
            return_response=False,
        )
        return response

    def get_delete_persons_schema(self) -> JsonType:
        """GETs the current schema from CIM

        The schema can change depending on the installation.

        It may be a good idea to setup some sort of daily(?)cron job that
        checks whether the schema has changed lately so you don't update
        using an outdated schema.
        """
        url = self.endpoints.get_delete_persons_schema()
        response = self.get(
            url,
            auth=self.auth,
            return_response=False,
        )
        return response

    def upsert_persons(
        self,
        persondata: Union[List[PersonBase], PersonBase, PersonList],
    ) -> Optional[Tuple[str, bytes]]:
        """Update one or more persons

        Note that updating more than one person at a time may require parsing
        the content of the response.
        The reason behind this is that you will not get a 404 if at least one
        person in the list was found.
        Note also that the response will not contain info about failing to find
        someone. You will have to compare with what you posted.
        """
        url = self.endpoints.upsert_persons()

        if isinstance(persondata, PersonList):
            persondata = persondata.persons
        elif isinstance(persondata, PersonBase):
            persondata = [persondata]
        elif not isinstance(persondata, list):
            raise TypeError("persondata must be List of Persons or Person")

        data = [_serialise_upsert_person(x) for x in persondata]
        headers = {"Content-Type": "application/json"}
        response = self.post(url, json=data, auth=self.auth, headers=headers)
        if response.status_code == 200:
            logger.debug("import success, response was: %s", response.content)
            return "Import success", response.content
        response.raise_for_status()
        return None

    def delete_persons(
        self,
        person_ids: List[str],
    ) -> Union[Tuple[str, bytes], None]:
        """Convenience method for deletion

        The request body should contain an array of unique identifiers
        for the users or contacts you want to delete. You can find out
        which property is used as a unique identifier in the installation
        by generating a JSON schema.

        :param person_ids: The person or people we want to delete from CIM
        :return: String describing status or None if not found
        """
        url = self.endpoints.delete_persons()
        headers = {"Content-Type": "application/json"}
        response = self.post(url, json=person_ids, auth=self.auth, headers=headers)

        if (
            response.status_code == 404
            and response.text == "No person found: No deletion"
        ):
            logger.warning(
                "No persons were deleted, response was: %s", response.content
            )
            return None

        if response.status_code == 200:
            logger.debug("Delete success, response was: %s", response.content)
            return "Delete success", response.content

        response.raise_for_status()
        return None

    def get_import_organisations_schema(self) -> JsonType:
        """GETs the current schema from CIM

        The schema can change depending on the installation.

        It may be a good idea to setup some sort of daily(?)cron job that
        checks whether the schema has changed lately so you don't update
        using an outdated schema.
        """
        url = self.endpoints.get_import_organisations_schema()
        response = self.get(
            url,
            auth=self.auth,
            return_response=False,
        )
        return response

    def import_organisations(
        self, organisations: Union[Iterable[Organisation], OrganisationList]
    ) -> Union[bool, OrganisationImportResult]:
        """Import organisation tree into CIM

        :param organisations: The list of organisations to import
        :return: OrganisationImportResult or ['No changes'] if no changes

        From CIM API documentation:

        NB! All imported organization units must be present at all times in the
        import to prevent deletion.

        NB! Note that the “parent” property of each organization unit in the
        import data should contain a valid organization key or be left empty. If
        “parent” contains a non-valid key at first-time import, the unit and its
        subordinate units will not be imported. If “parent” contains a non-valid
        key when updates are made to an existing unit, the unit and all its
        subordinate units will be deleted.

        Result:

        The import resulted in several organisations/departments being inserted:

            {
              "insert":["ORG-123","ORG-124"],
              "update":["ORG-121","ORG-21","ORG-211"],
              "delete":["ORG-121w"]
            }

        NB! If an organisation had been deleted as a result of a previous
        import, and was afterwards restored via import, the result will be sent
        as “update”.

        No changes were made as a result of the import:

            ["No changes"]
        """
        if not isinstance(organisations, OrganisationList):
            organisations = OrganisationList(__root__=list(organisations))  # noqa

        # Api doesn't give any feedback on unknown parents, log them
        unknown_parents = _find_unknown_parents(organisations.__root__)
        if unknown_parents:
            logger.warning("%s keys not found", len(unknown_parents))
            logger.debug("Unknown keys: %s", unknown_parents)

        url = self.endpoints.import_organisations()
        result: JsonType = self.post(
            url,
            json=organisations.dict()["__root__"],
            auth=self.auth,
            return_response=False,
        )

        if isinstance(result, dict):
            return OrganisationImportResult.from_dict(result)

        if result != ["No changes"]:
            logger.warning("Unrecognized result: %s", result)

        return result


def _find_unknown_parents(organisations: List[Organisation]) -> Set[str]:
    keys = {x.key for x in organisations}
    parent_keys = set(
        filter(lambda x: bool(x), map(lambda x: x.parent_key, organisations))
    )

    if not parent_keys.issubset(keys):
        return parent_keys - keys

    return set()


def get_client(config_dict: JsonType) -> CimClient:
    """
    Get a CimClient from configuration.
    """
    return CimClient(**config_dict)