Skip to content
Back to top

# Source code for slixmpp.plugins.xep_0191.blocking

    # Slixmpp: The Slick XMPP Library
    # Copyright (C) 2012 Nathanael C. Fritz, Lance J.T. Stout
    # This file is part of Slixmpp.
    # See the file LICENSE for copying permission.
    import logging

    from asyncio import Future

    from slixmpp.stanza import Iq
    from slixmpp.plugins import BasePlugin
    from slixmpp.xmlstream.handler import Callback
    from slixmpp.xmlstream.matcher import StanzaPath
    from slixmpp.xmlstream import register_stanza_plugin, JID
    from slixmpp.plugins.xep_0191 import stanza, Block, Unblock, BlockList, BlockItem

    log = logging.getLogger(__name__)

    BlockedJIDs = (
        JID |
        set[JID] |
        list[JID]
    )

    class XEP_0191(BasePlugin):

        name = 'xep_0191'
        description = 'XEP-0191: Blocking Command'
        dependencies = {'xep_0030'}
        stanza = stanza

        def plugin_init(self):
            register_stanza_plugin(Iq, BlockList)
            register_stanza_plugin(Iq, Block)
            register_stanza_plugin(Iq, Unblock)
            register_stanza_plugin(BlockList, BlockItem, iterable=True)
            register_stanza_plugin(Block, BlockItem, iterable=True)
            register_stanza_plugin(Unblock, BlockItem, iterable=True)

            self.xmpp.register_handler(
                    Callback('Blocked Contact',
                        StanzaPath('iq@type=set/block'),
                        self._handle_blocked))

            self.xmpp.register_handler(
                    Callback('Unblocked Contact',
                        StanzaPath('iq@type=set/unblock'),
                        self._handle_unblocked))

        def plugin_end(self):
            self.xmpp.remove_handler('Blocked Contact')
            self.xmpp.remove_handler('Unblocked Contact')

        def get_blocked(self, ifrom: JID | None = None, **iqkwargs) -> Future:
            """Get the iq containing the blocklist."""
            iq = self.xmpp.make_iq_get(ifrom=ifrom)
            iq.enable('blocklist')
            return iq.send(**iqkwargs)

        async def get_blocked_jids(self, ifrom: JID | None = None, **iqkwargs) -> set[JID]:
            """Get the list of blocked JIDs."""
            iq = self.xmpp.make_iq_get(ifrom=ifrom)
            iq.enable('blocklist')
            result = await iq.send(**iqkwargs)
            return {JID(item['jid']) for item in result['blocklist']}

        def block(self, jids: BlockedJIDs,
                  ifrom: JID | None = None, **iqkwargs) -> Future:
            """Block a JID or a list of JIDs.

            :param jids: JID(s) to block.
            """
            iq = self.xmpp.make_iq_set(ifrom=ifrom)
            if not isinstance(jids, (set, list)):
                jids = [jids]

            for jid in jids:
                item = BlockItem(parent=iq['block'])
                item['jid'] = jid

            return iq.send(**iqkwargs)

        def unblock(self, jids: BlockedJIDs, ifrom: JID | None = None, **iqkwargs) -> Future:
            """Unblock a JID or a list of JIDs.

            :param jids: JID(s) to unblock.
            """
            if jids is None:
                raise ValueError("jids cannot be empty.")
            iq = self.xmpp.make_iq_set(ifrom=ifrom)

            if not isinstance(jids, (set, list)):
                jids = [jids]

            for jid in jids:
                item = BlockItem(parent=iq['unblock'])
                item['jid'] = jid

            return iq.send(**iqkwargs)

        def _handle_blocked(self, iq):
            self.xmpp.event('blocked', iq)

        def _handle_unblocked(self, iq):
            self.xmpp.event('unblocked', iq)