Source code for mailmanclient.asynclient

# Copyright (C) 2021-2023 by the Free Software Foundation, Inc.
# This file is part of mailman.client.
# mailman.client is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by the
# Free Software Foundation, version 3 of the License.
# mailman.client is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
# or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public
# License for more details.
# You should have received a copy of the GNU Lesser General Public License
# along with mailman.client.  If not, see <>.
"""Async client for Mailman Core 3.1 API.

AsyncClient provides a thin Python API over Mailman Core's HTTP API. It is
currently is very early stages and supports on read operations. Some of the
read operations might be missinng as well.

To start using the client, you need an async http library. httpx_ is officially
supported one, but making some other client work with it is pretty easy.

    >>> import httpx
    >>> conn = httpx.AsyncClient()
    >>> from mailmanclient.asynclient import AsyncClient
    >>> client = AsyncClient(conn, 'http://localhost:8001/3.1',
    ...                      'restadmin', 'restpass')

You will need an event loop to actually run the client. You can reuse your
existing one by calling ``await`` on the client methods, or you can create a
new one using the standard library ``asyncio`` module.

    >>> import asyncio
    >>> domains =

.. _httpx:


__all__ = [

from typing import List, Mapping, Any
from mailmanclient.restobjects.utils import list_of_objects
from mailmanclient.restobjects.types import HTTPClientProto
from mailmanclient.restbase.async_connection import Connection
from mailmanclient.asyncobjects.domain import Domain
from mailmanclient.asyncobjects.mailinglist import MailingList
from mailmanclient.asyncobjects.user import User
from mailmanclient.asyncobjects.address import Address
from mailmanclient.asyncobjects.member import Member

JSON_CONTENT_TYPE = 'application/json'

[docs] class AsyncClient: """Provide an Idiomatic API for Mailman Core. It requires an HTTP client instance as the first argument. You can use any client which has a ``.request()`` method and accepts named parameters ``url``, ``path``, ``auth``, ``method`` and ``data``. ``data`` is supposed to be a dictionary of parameters to be passed to the HTTP request and the rest are string parameters with their usual meaning. The parameters are based off on httpx python library. :param client: Http client object with an async request method. :param base_url: Base URL to Core's API. :param user: Core admin username. :param password: Core admin password. """ def __init__( self, client: HTTPClientProto, base_url: str, user: str, password: str, ) -> None: self.client = client self.connection = Connection(self.client, base_url, user, password)
[docs] async def domains(self) -> List[Domain]: """Get all domains. ``/<api>/domains`` """ response, content = await'domains') return list_of_objects(Domain, content, self.connection)
[docs] async def system(self) -> Mapping[str, Any]: """Get the Mailman system information. ``/<api>/system`` """ response, content = await'system') return content
[docs] async def lists(self) -> List[MailingList]: """Get a list of MailingLists ``/<api>/lists`` """ response, content = await'lists') return list_of_objects(MailingList, content, self.connection)
[docs] async def members(self) -> List[Member]: """All the Members ``/<api>/members`` """ response, content = await'members') return list_of_objects(Member, content, self.connection)
[docs] async def users(self) -> List[User]: """All the users in Mailman Core ``/<api>/users`` """ response, content = await'users') return list_of_objects(User, content, self.connection)
[docs] async def addresses(self) -> List[Address]: """All the addresses in Mailman ``/<api>/address`` """ response, content = await'addresses') return list_of_objects(Address, content, self.connection)
[docs] async def find_members( self, list_id: str = None, subscriber: str = None, role: str = None, moderation_action: str = None, delivery_status: str = None, delivery_mode: str = None, ) -> List[Member]: """Find members. ``/<api>/members/find`` :param list_id: Mailinglist id. :param subscriber: Email or user_id or partial search string. :param role: Membership role. One of 'owner', 'member', 'nonmember' or 'moderator'. :param moderation_action: One of the moderation action from 'defer', 'accept', 'discard', 'reject', 'hold'. :param delivery_status: Delivery status of the Member. It can be one among 'enabled', 'by_user', 'by_moderator' or 'by_bounces'. :param delivery_mode: Delivery mode of the member. It can be one between 'plaintext_digests', 'mime_digests', 'regular'. """ data = dict(list_id=list_id, subscriber=subscriber, role=role, moderation_action=moderation_action, delivery_status=delivery_status, delivery_mode=delivery_mode) # Skip parameters that have None value. # TODO: Handle parameters that can have None value. data = {key: value for key, value in data.items() if value is not None} response, content = await 'members/find', data=data, method='GET') return list_of_objects(Member, content, self.connection)