py-sympa-soap/sympasoap/client.py

from zeep.client import Client as ZeepClient, Settings as ZeepSettings
from zeep.exceptions import Fault

from .lists import MailingList, MLUser
from .constants import SUBTOPICS, TOPICS, TEMPLATES


class Client:
    def __init__(self, sympa_url: str):
        self.sympa_url = sympa_url
        self.zeep = ZeepClient(sympa_url + "/wsdl", settings=ZeepSettings(strict=False))
        self.cookie = None

    def login(self, email: str, password: str) -> None:
        """
        Login into the API. Set a cookie for future connexions.
        """
        result = self.zeep.service.login(email, password)
        element = result._raw_elements[0]
        self.cookie = element.text
        self.zeep.settings.extra_http_headers = [("Cookie", f"sympa_session={element.text}")]
        if self.check_cookie() != email:
            # FIXME Better exception
            raise Exception("Unknown error: given cookie is invalid")
        self.email = email
        print("Successfully authenticated!")

    def logout(self):
        """
        Clear cookie
        """
        self.cookie = None
        self.email = None
        self.zeep.settings.extra_http_headers = []

    def check_cookie(self) -> str:
        """
        From the current cookie, retrieve the email address.
        """
        result = self.zeep.service.checkCookie()
        element = result._raw_elements[0]
        return element.text

    def is_subscriber(self, email: str, mailing_list: str, function: str = "subscriber") -> bool:
        """
        Check if the given `email` is a member of type `function` in the `mailing_list`.
        The function parameter is one between subscriber, editor or owner.
        """
        if function not in ["subscriber", "editor", "owner"]:
            raise ValueError("function of a mailing list member must be subscriber, editor or owner.")
        result = self.zeep.service.amI(mailing_list, function, email)
        element = result._raw_elements[0]
        return element.text == "true"

    def get_subscribers(self, mailing_list: str, emails_only: bool = True) -> list:
        """
        Get the list of all subscribers of a list, including the administrators and the editors.
        If emails_only == True, retrieve the list of email addresses only.
        Else, retrieve MLUser object, with the name of the user and the role.
        """
        if not emails_only:
            users = list()
            elements = self.zeep.service.fullReview(mailing_list)
            for element in elements:
                children = element.getchildren()
                kwargs = dict(mailing_list=mailing_list)
                for child in children:
                    tag = child.tag
                    if "gecos" in tag:
                        kwargs["name"] = child.text
                    elif "email" in tag:
                        kwargs["email"] = child.text
                    elif "isSubscriber" in tag:
                        kwargs["subscriber"] = child.text == "true"
                    elif "isEditor" in tag:
                        kwargs["editor"] = child.text == "true"
                    elif "isOwner" in tag:
                        kwargs["owner"] = child.text == "true"
                    else:
                        print("Unknown child tag:", tag)
                user = MLUser(**kwargs)
                users.append(user)
            return users
        return self.zeep.service.review(mailing_list)

    def lists(self, topic: str, subtopic: str) -> list:
        """
        Get all the (visible) lists that match the given topic and the given subtopic.
        See TOPICS and SUBTOPICS for valid topics and subtopics.
        """
        if topic not in TOPICS:
            raise ValueError(f"'{topic}' is not a valid topic.")
        if subtopic and f"{topic}/{subtopic}" not in SUBTOPICS:
            raise ValueError(f"'{topic}/{subtopic}' is not a valid subtopic.")
        result = self.zeep.service.lists(topic, subtopic)._value_1
        if result is None:
            return list()
        lists = list()
        for list_info in result:
            split = list_info.split(';')
            kwargs = dict()
            for data in split:
                key, value = data.split("=", 2)
                if key == "listAddress":
                    key = "list_address"
                kwargs[key] = value
            ml = MailingList(**kwargs)
            lists.append(ml)
        return lists

    def all_lists(self) -> list:
        """
        Retrieve all lists.
        """
        elem = self.zeep.service.complexLists()._raw_elements[0]
        lists = list()
        for list_info in elem.getchildren():
            kwargs = dict()
            for child in list_info.getchildren():
                if "listAddress" in child.tag:
                    key = "list_address"
                elif "subject" in child.tag:
                    key = "subject"
                elif "homepage" in child.tag:
                    key = "homepage"
                else:
                    raise ValueError(f"Tag {child.tag} is unknown")
                kwargs[key] = child.text
            ml = MailingList(**kwargs)
            lists.append(ml)
        return lists

    def create_list(self, list_name: str, subject: str, template: str, description: str, topic: str,
                    use_custom_template: bool = False, raise_error: bool = True) -> bool:
        """
        Create a new mailing-list.
        """
        if topic not in TOPICS and topic not in SUBTOPICS:
            raise ValueError(f"Topic '{topic}' does not exist.")
        if template not in TEMPLATES and not use_custom_template:
            raise ValueError(f"Template '{template}' does not exist.")
        try:
            result = self.zeep.service.createList(list_name, subject, template, description, topic)
            return result._raw_elements[0].text == "true"
        except Fault as e:
            if raise_error:
                raise Fault(f"An unknown error occured while creating the list {list_name}. "
                            f"Maybe the list already exists?", e)
            else:
                return False

    def delete_list(self, list_name: str, raise_error: bool = False) -> bool:
        """
        Close a mailing list.
        Warning: the list is not deleted in order to keep the history. Please use the web interface to fully
        delete the list.
        Well, the main reason is that the API does not provide a delete method.
        """
        try:
            result = self.zeep.service.closeList(list_name)
            return result._raw_elements[0].text == "true"
        except Fault as e:
            if raise_error:
                raise Fault(f"An unknown error occured while deleting the list {list_name}. "
                            f"Maybe the list did not exist?", e)
            else:
                return False