XEP-0045: Types, visibility, and documentation

- Make all handlers private (_-prefixed)
- Reorder methods in a more thematic order
- Add docstrings to public methods
- Add types where they were missing
- Create new Literal types for closed enums
- Make join_muc a wrapper around join_muc_wait and return a Future
- Deprecate the current join_muc API
- Fix some mypy issues
This commit is contained in:
mathieui 2021-02-06 18:05:59 +01:00
parent b9e479f213
commit bc04da256a
2 changed files with 314 additions and 136 deletions

View file

@ -9,6 +9,7 @@ import asyncio
import logging import logging
from datetime import datetime from datetime import datetime
from typing import ( from typing import (
Any,
Dict, Dict,
List, List,
Tuple, Tuple,
@ -29,6 +30,7 @@ from slixmpp.xmlstream.matcher.stanzapath import StanzaPath
from slixmpp.xmlstream.matcher.xmlmask import MatchXMLMask from slixmpp.xmlstream.matcher.xmlmask import MatchXMLMask
from slixmpp.exceptions import IqError, IqTimeout, PresenceError from slixmpp.exceptions import IqError, IqTimeout, PresenceError
from slixmpp.plugins.xep_0004 import Form
from slixmpp.plugins.xep_0045 import stanza from slixmpp.plugins.xep_0045 import stanza
from slixmpp.plugins.xep_0045.stanza import ( from slixmpp.plugins.xep_0045.stanza import (
MUCInvite, MUCInvite,
@ -45,6 +47,13 @@ from slixmpp.plugins.xep_0045.stanza import (
MUCActor, MUCActor,
MUCUserItem, MUCUserItem,
) )
from slixmpp.types import (
MucRole,
MucAffiliation,
MucRoomItem,
MucRoomItemKeys,
PresenceArgs,
)
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@ -56,7 +65,7 @@ ROLES = ('moderator', 'participant', 'visitor', 'none')
class XEP_0045(BasePlugin): class XEP_0045(BasePlugin):
""" """
Implements XEP-0045 Multi-User Chat XEP-0045 Multi-User Chat
""" """
name = 'xep_0045' name = 'xep_0045'
@ -64,6 +73,9 @@ class XEP_0045(BasePlugin):
dependencies = {'xep_0030', 'xep_0004'} dependencies = {'xep_0030', 'xep_0004'}
stanza = stanza stanza = stanza
rooms: Dict[JID, Dict[str, MucRoomItem]]
our_nicks: Dict[JID, str]
def plugin_init(self): def plugin_init(self):
self.rooms = {} self.rooms = {}
self.our_nicks = {} self.our_nicks = {}
@ -82,6 +94,7 @@ class XEP_0045(BasePlugin):
register_stanza_plugin(Iq, MUCAdminQuery) register_stanza_plugin(Iq, MUCAdminQuery)
register_stanza_plugin(Iq, MUCOwnerQuery) register_stanza_plugin(Iq, MUCOwnerQuery)
register_stanza_plugin(MUCOwnerQuery, MUCOwnerDestroy) register_stanza_plugin(MUCOwnerQuery, MUCOwnerDestroy)
register_stanza_plugin(MUCOwnerQuery, Form)
register_stanza_plugin(MUCAdminQuery, MUCAdminItem, iterable=True) register_stanza_plugin(MUCAdminQuery, MUCAdminItem, iterable=True)
# Register handlers # Register handlers
@ -89,7 +102,7 @@ class XEP_0045(BasePlugin):
Callback( Callback(
'MUCPresence', 'MUCPresence',
StanzaPath("presence/muc"), StanzaPath("presence/muc"),
self.handle_groupchat_presence, self._handle_groupchat_presence,
)) ))
# <x xmlns="http://jabber.org/protocol/muc"/> is only used in # <x xmlns="http://jabber.org/protocol/muc"/> is only used in
# presence when joining on the client side, and for errors on # presence when joining on the client side, and for errors on
@ -99,7 +112,7 @@ class XEP_0045(BasePlugin):
Callback( Callback(
'MUCPresenceJoin', 'MUCPresenceJoin',
StanzaPath("presence/muc_join"), StanzaPath("presence/muc_join"),
self.handle_groupchat_join, self._handle_groupchat_join,
)) ))
self.xmpp.register_handler( self.xmpp.register_handler(
Callback( Callback(
@ -113,37 +126,37 @@ class XEP_0045(BasePlugin):
Callback( Callback(
'MUCError', 'MUCError',
MatchXMLMask("<message xmlns='%s' type='error'><error/></message>" % self.xmpp.default_ns), MatchXMLMask("<message xmlns='%s' type='error'><error/></message>" % self.xmpp.default_ns),
self.handle_groupchat_error_message self._handle_groupchat_error_message
)) ))
self.xmpp.register_handler( self.xmpp.register_handler(
Callback( Callback(
'MUCMessage', 'MUCMessage',
MatchXMLMask("<message xmlns='%s' type='groupchat'><body/></message>" % self.xmpp.default_ns), MatchXMLMask("<message xmlns='%s' type='groupchat'><body/></message>" % self.xmpp.default_ns),
self.handle_groupchat_message self._handle_groupchat_message
)) ))
self.xmpp.register_handler( self.xmpp.register_handler(
Callback( Callback(
'MUCSubject', 'MUCSubject',
MatchXMLMask("<message xmlns='%s' type='groupchat'><subject/></message>" % self.xmpp.default_ns), MatchXMLMask("<message xmlns='%s' type='groupchat'><subject/></message>" % self.xmpp.default_ns),
self.handle_groupchat_subject self._handle_groupchat_subject
)) ))
self.xmpp.register_handler( self.xmpp.register_handler(
Callback( Callback(
'MUCConfig', 'MUCConfig',
StanzaPath('message/muc/status'), StanzaPath('message/muc/status'),
self.handle_config_change self._handle_config_change
)) ))
self.xmpp.register_handler( self.xmpp.register_handler(
Callback( Callback(
'MUCInvite', 'MUCInvite',
StanzaPath('message/muc/invite'), StanzaPath('message/muc/invite'),
self.handle_groupchat_invite self._handle_groupchat_invite
)) ))
self.xmpp.register_handler( self.xmpp.register_handler(
Callback( Callback(
'MUCDecline', 'MUCDecline',
StanzaPath('message/muc/decline'), StanzaPath('message/muc/decline'),
self.handle_groupchat_decline self._handle_groupchat_decline
)) ))
def plugin_end(self): def plugin_end(self):
@ -152,7 +165,7 @@ class XEP_0045(BasePlugin):
def session_bind(self, jid): def session_bind(self, jid):
self.xmpp.plugin['xep_0030'].add_feature(stanza.NS) self.xmpp.plugin['xep_0030'].add_feature(stanza.NS)
def handle_groupchat_invite(self, inv): def _handle_groupchat_invite(self, inv: Message):
""" Handle an invite into a muc. """ """ Handle an invite into a muc. """
if self.xmpp.is_component: if self.xmpp.is_component:
self.xmpp.event('groupchat_invite', inv) self.xmpp.event('groupchat_invite', inv)
@ -160,7 +173,7 @@ class XEP_0045(BasePlugin):
if inv['from'] not in self.rooms.keys(): if inv['from'] not in self.rooms.keys():
self.xmpp.event("groupchat_invite", inv) self.xmpp.event("groupchat_invite", inv)
def handle_groupchat_decline(self, decl): def _handle_groupchat_decline(self, decl: Message):
"""Handle an invitation decline.""" """Handle an invitation decline."""
if self.xmpp.is_component: if self.xmpp.is_component:
self.xmpp.event('groupchat_invite', decl) self.xmpp.event('groupchat_invite', decl)
@ -168,12 +181,12 @@ class XEP_0045(BasePlugin):
if decl['from'] in self.room.keys(): if decl['from'] in self.room.keys():
self.xmpp.event('groupchat_decline', decl) self.xmpp.event('groupchat_decline', decl)
def handle_config_change(self, msg): def _handle_config_change(self, msg: Message):
"""Handle a MUC configuration change (with status code).""" """Handle a MUC configuration change (with status code)."""
self.xmpp.event('groupchat_config_status', msg) self.xmpp.event('groupchat_config_status', msg)
self.xmpp.event('muc::%s::config_status' % msg['from'].bare , msg) self.xmpp.event('muc::%s::config_status' % msg['from'].bare , msg)
def client_handle_presence(self, pr: Presence): def _client_handle_presence(self, pr: Presence):
"""As a client, handle a presence stanza""" """As a client, handle a presence stanza"""
got_offline = False got_offline = False
got_online = False got_online = False
@ -206,61 +219,47 @@ class XEP_0045(BasePlugin):
"""Generate MUC presence error events""" """Generate MUC presence error events"""
self.xmpp.event("muc::%s::presence-error" % pr['from'].bare, pr) self.xmpp.event("muc::%s::presence-error" % pr['from'].bare, pr)
def handle_groupchat_presence(self, pr: Presence): def _handle_groupchat_presence(self, pr: Presence):
""" Handle a presence in a muc.""" """ Handle a presence in a muc."""
if self.xmpp.is_component: if self.xmpp.is_component:
self.xmpp.event('groupchat_presence', pr) self.xmpp.event('groupchat_presence', pr)
else: else:
self.client_handle_presence(pr) self._client_handle_presence(pr)
def handle_groupchat_join(self, pr: Presence): def _handle_groupchat_join(self, pr: Presence):
"""Received a join presence (as a component)""" """Received a join presence (as a component)"""
self.xmpp.event('groupchat_join', pr) self.xmpp.event('groupchat_join', pr)
def handle_groupchat_message(self, msg: Message) -> None: def _handle_groupchat_message(self, msg: Message):
""" Handle a message event in a muc. """ Handle a message event in a muc.
""" """
self.xmpp.event('groupchat_message', msg) self.xmpp.event('groupchat_message', msg)
self.xmpp.event("muc::%s::message" % msg['from'].bare, msg) self.xmpp.event("muc::%s::message" % msg['from'].bare, msg)
def handle_groupchat_error_message(self, msg): def _handle_groupchat_error_message(self, msg: Message):
""" Handle a message error event in a muc. """ Handle a message error event in a muc.
""" """
self.xmpp.event('groupchat_message_error', msg) self.xmpp.event('groupchat_message_error', msg)
self.xmpp.event("muc::%s::message_error" % msg['from'].bare, msg) self.xmpp.event("muc::%s::message_error" % msg['from'].bare, msg)
def handle_groupchat_subject(self, msg: Message) -> None: def _handle_groupchat_subject(self, msg: Message):
""" Handle a message coming from a muc indicating """ Handle a message coming from a muc indicating
a change of subject (or announcing it when joining the room) a change of subject (or announcing it when joining the room)
""" """
# See poezio#3452. A message containing subject _and_ (body or thread) # See poezio#3452. A message containing subject _and_ (body or thread)
# is not a subject change. # is not a subject change.
if msg['body'] or msg['thread']: if msg['body'] or msg['thread']:
return None return
self.xmpp.event('groupchat_subject', msg) self.xmpp.event('groupchat_subject', msg)
def jid_in_room(self, room: JID, jid: JID) -> bool:
for nick in self.rooms[room]:
entry = self.rooms[room][nick]
if entry is not None and entry['jid'].full == jid:
return True
return False
def get_nick(self, room: JID, jid: JID) -> Optional[str]:
for nick in self.rooms[room]:
entry = self.rooms[room][nick]
if entry is not None and entry['jid'].full == jid:
return nick
return None
async def join_muc_wait(self, room: JID, nick: str, *, async def join_muc_wait(self, room: JID, nick: str, *,
password: Optional[str] = None, password: Optional[str] = None,
maxchars: Optional[int] = None, maxchars: Optional[int] = None,
maxstanzas: Optional[int] = None, maxstanzas: Optional[int] = None,
seconds: Optional[int] = None, seconds: Optional[int] = None,
since: Optional[datetime] = None, since: Optional[datetime] = None,
presence_options: Optional[Dict[str, str]] = None, presence_options: Optional[PresenceArgs] = None,
timeout: Optional[int] = None) -> Presence: timeout: Optional[int] = None) -> Presence:
""" """
Try to join a MUC and block until we are joined or get an error. Try to join a MUC and block until we are joined or get an error.
@ -268,6 +267,8 @@ class XEP_0045(BasePlugin):
Only one of {maxchars, maxstanzas, seconds, since} will be used, in Only one of {maxchars, maxstanzas, seconds, since} will be used, in
that order. that order.
.. versionadded:: 1.8.0
:param password: The optional room password. :param password: The optional room password.
:param maxchars: Max number of characters to return from history. :param maxchars: Max number of characters to return from history.
:param maxstanzas: Max number of stanzas to return from history. :param maxstanzas: Max number of stanzas to return from history.
@ -303,7 +304,7 @@ class XEP_0045(BasePlugin):
self.our_nicks[room] = nick self.our_nicks[room] = nick
stanza.send() stanza.send()
future = asyncio.Future() future: asyncio.Future = asyncio.Future()
context1 = self.xmpp.event_handler("muc::%s::self-presence" % room, future.set_result) context1 = self.xmpp.event_handler("muc::%s::self-presence" % room, future.set_result)
context2 = self.xmpp.event_handler("muc::%s::presence-error" % room, future.set_result) context2 = self.xmpp.event_handler("muc::%s::presence-error" % room, future.set_result)
with context1, context2: with context1, context2:
@ -321,35 +322,118 @@ class XEP_0045(BasePlugin):
return pres return pres
def join_muc(self, room: JID, nick: str, maxhistory="0", password='', def join_muc(self, room: JID, nick: str, maxhistory="0", password='',
pstatus='', pshow='', pfrom=''): pstatus='', pshow='', pfrom='') -> asyncio.Future:
""" Join the specified room, requesting 'maxhistory' lines of history. """ Join the specified room, requesting 'maxhistory' lines of history.
.. deprecated:: 1.8.0
:meth:`join_muc_wait` will replace this old API starting from version
1.9.0.
""" """
stanza = self.xmpp.make_presence( presence_options = PresenceArgs(
pto="%s/%s" % (room, nick), pstatus=pstatus, pshow=pshow,
pshow=pshow, pfrom=pfrom pstatus=pstatus,
pfrom=pfrom,
) )
stanza.enable('muc_join') maxchars, maxstanzas = None, None
if password:
stanza['muc_join']['password'] = password
if maxhistory: if maxhistory:
if maxhistory == "0": if maxhistory == "0":
stanza['muc_join']['history']['maxchars'] = '0' maxchars = 9
else: else:
stanza['muc_join']['history']['maxstanzas'] = str(maxhistory) maxstanzas = int(maxhistory)
self.xmpp.send(stanza) return asyncio.ensure_future(
self.rooms[room] = {} self.join_muc_wait(
self.our_nicks[room] = nick room=room,
nick=nick,
password=password,
presence_options=presence_options,
maxchars=maxchars,
maxstanzas=maxstanzas,
),
loop=self.xmpp.loop,
)
def leave_muc(self, room: JID, nick: str, msg: str = '', pfrom: Optional[JID] = None):
""" Leave the specified room.
:param room: Room to leave.
:param nick: Your nickname.
:param msg: Presence status to use.
"""
if msg:
self.xmpp.send_presence(
pshow='unavailable',
pto="%s/%s" % (room, nick),
pstatus=msg,
pfrom=pfrom
)
else:
self.xmpp.send_presence(
pshow='unavailable',
pto="%s/%s" % (room, nick),
pfrom=pfrom
)
del self.rooms[room]
def set_subject(self, room: JID, subject: str, *, mfrom: Optional[JID] = None): def set_subject(self, room: JID, subject: str, *, mfrom: Optional[JID] = None):
"""Set a rooms subject.""" """Set a rooms subject.
:param room: JID of the room.
:param subject: Room subject to set.
"""
msg = self.xmpp.make_message(room, mfrom=mfrom) msg = self.xmpp.make_message(room, mfrom=mfrom)
msg['type'] = 'groupchat' msg['type'] = 'groupchat'
msg['subject'] = subject msg['subject'] = subject
msg.send() msg.send()
async def destroy(self, room: JID, reason='', altroom='', *, async def get_room_config(self, room: JID, ifrom: Optional[JID] = None,
**iqkwargs) -> Form:
"""Get the room config form in 0004 plugin format.
:param room: Room to get the config form from.
:raises ValueError: When the form is not found.
:returns: A form object.
"""
iq = self.xmpp.make_iq_get(stanza.NS_OWNER, ito=room, ifrom=ifrom)
result = await iq.send(**iqkwargs)
form = result['mucowner_query'].get_plugin('form', check=True)
if form is None:
raise ValueError("Configuration form not found")
return form
async def set_room_config(self, room: JID, config: Form, *,
ifrom: Optional[JID] = None, **iqkwargs): ifrom: Optional[JID] = None, **iqkwargs):
"""Destroy a room.""" """Send a room config form.
:param room: Room to send the form to.
:param config: A filled room form.
"""
query = MUCOwnerQuery()
config['type'] = 'submit'
query.append(config)
iq = self.xmpp.make_iq_set(query, ito=room, ifrom=ifrom)
await iq.send(**iqkwargs)
async def cancel_config(self, room: JID, *,
ifrom: Optional[JID] = None, **iqkwargs):
"""Cancel a requested config form.
:param room: Room to cancel the form for.
"""
query = MUCOwnerQuery()
query['form']['type'] = 'cancel'
iq = self.xmpp.make_iq_set(query, ito=room, ifrom=ifrom)
await iq.send(**iqkwargs)
async def destroy(self, room: JID, reason: str = '', altroom: Optional[JID] = None, *,
ifrom: Optional[JID] = None, **iqkwargs):
"""Destroy a room.
:param room: Room JID to destroy.
:param reason: Reason for destroying the room.
:param altroom: An alternate room that users should join.
"""
iq = self.xmpp.make_iq_set(ifrom=ifrom, ito=room) iq = self.xmpp.make_iq_set(ifrom=ifrom, ito=room)
iq.enable('mucowner_query') iq.enable('mucowner_query')
iq['mucowner_query'].enable('destroy') iq['mucowner_query'].enable('destroy')
@ -359,10 +443,17 @@ class XEP_0045(BasePlugin):
iq['mucowner_query']['destroy']['reason'] = reason iq['mucowner_query']['destroy']['reason'] = reason
await iq.send(**iqkwargs) await iq.send(**iqkwargs)
async def set_affiliation(self, room: JID, affiliation: str, *, jid: Optional[JID] = None, async def set_affiliation(self, room: JID, affiliation: MucAffiliation, *,
jid: Optional[JID] = None,
nick: Optional[str] = None, reason: str = '', nick: Optional[str] = None, reason: str = '',
ifrom: Optional[JID] = None, **iqkwargs): ifrom: Optional[JID] = None, **iqkwargs):
""" Change room affiliation.""" """ Change room affiliation for a JID or nickname.
:param room: Room to modify.
:param affiliation: Affiliation to set.
:param jid: User JID to use in the set operation.
:param reason: Reason for the affiliation change.
"""
if affiliation not in AFFILIATIONS: if affiliation not in AFFILIATIONS:
raise ValueError('%s is not a valid affiliation' % affiliation) raise ValueError('%s is not a valid affiliation' % affiliation)
if not any((jid, nick)): if not any((jid, nick)):
@ -377,12 +468,45 @@ class XEP_0045(BasePlugin):
iq['mucadmin_query']['item']['reason'] = reason iq['mucadmin_query']['item']['reason'] = reason
await iq.send(**iqkwargs) await iq.send(**iqkwargs)
async def set_role(self, room: JID, nick: str, role: str, *, async def get_affiliation_list(self, room: JID, affiliation: MucAffiliation, *,
ifrom: Optional[JID] = None, **iqkwargs) -> List[JID]:
"""Get a list of JIDs with the specified affiliation
:param room: Room to get affiliations from.
:param affiliation: The affiliation to list.
"""
iq = self.xmpp.make_iq_get(stanza.NS_ADMIN, ito=room, ifrom=ifrom)
iq['mucadmin_query']['item']['affiliation'] = affiliation
result = await iq.send(**iqkwargs)
return [item['jid'] for item in result['mucadmin_query']]
async def send_affiliation_list(self, room: JID,
affiliations: List[Tuple[JID, MucAffiliation]], *,
ifrom: Optional[JID] = None, **iqkwargs):
"""Send an affiliation delta list.
:param room: Room to send the affiliations to.
:param affiliations: List of couples (jid, affiliation) to set.
"""
iq = self.xmpp.make_iq_set(ito=room, ifrom=ifrom)
for jid, affiliation in affiliations:
item = MUCAdminItem()
item['jid'] = jid
item['affiliation'] = affiliation
iq['mucadmin_query'].append(item)
await iq.send(**iqkwargs)
async def set_role(self, room: JID, nick: str, role: MucRole, *,
reason: str = '', ifrom: Optional[JID] = None, **iqkwargs): reason: str = '', ifrom: Optional[JID] = None, **iqkwargs):
""" Change role property of a nick in a room. """ Change role property of a nick in a room.
Typically, roles are temporary (they last only as long as you are in the Typically, roles are temporary (they last only as long as you are in the
room), whereas affiliations are permanent (they last across groupchat room), whereas affiliations are permanent (they last across groupchat
sessions). sessions).
:param room: Room to modify.
:param nick: User nickname to use in the set operation.
:param role: Role to set.
:param reason: Reason for the role change.
""" """
if role not in ROLES: if role not in ROLES:
raise ValueError("Role %s does not exist" % role) raise ValueError("Role %s does not exist" % role)
@ -393,110 +517,127 @@ class XEP_0045(BasePlugin):
iq['mucadmin_query']['item']['reason'] = reason iq['mucadmin_query']['item']['reason'] = reason
await iq.send(**iqkwargs) await iq.send(**iqkwargs)
def invite(self, room: JID, jid: JID, reason: str = '', *, async def get_roles_list(self, room: JID, role: MucRole, *,
mfrom: Optional[JID] = None):
""" Invite a jid to a room."""
msg = self.xmpp.make_message(room, mfrom=mfrom)
msg['muc']['invite']['to'] = jid
if reason:
msg['muc']['invite']['reason'] = reason
self.xmpp.send(msg)
def decline(self, room: JID, jid: JID, reason: str = '', *,
mfrom: Optional[JID] = None):
"""Decline a mediated invitation."""
msg = self.xmpp.make_message(room, mfrom=mfrom)
msg['muc']['decline']['to'] = jid
if reason:
msg['muc']['decline']['reason'] = reason
self.xmpp.send(msg)
def leave_muc(self, room: JID, nick: str, msg='', pfrom=None):
""" Leave the specified room.
"""
if msg:
self.xmpp.send_presence(pshow='unavailable', pto="%s/%s" % (room, nick), pstatus=msg, pfrom=pfrom)
else:
self.xmpp.send_presence(pshow='unavailable', pto="%s/%s" % (room, nick), pfrom=pfrom)
del self.rooms[room]
async def get_room_config(self, room: JID, ifrom=''):
"""Get the room config form in 0004 plugin format """
iq = self.xmpp.make_iq_get(stanza.NS_OWNER, ito=room, ifrom=ifrom)
# For now, swallow errors to preserve existing API
result = await iq.send()
form = result.xml.find('{http://jabber.org/protocol/muc#owner}query/{jabber:x:data}x')
if form is None:
raise ValueError("Configuration form not found")
return self.xmpp.plugin['xep_0004'].build_form(form)
async def cancel_config(self, room: JID, *,
ifrom: Optional[JID] = None, **iqkwargs) -> Iq:
"""Cancel a requested config form"""
query = MUCOwnerQuery()
x = ET.Element('{jabber:x:data}x', type='cancel')
query.append(x)
iq = self.xmpp.make_iq_set(query, ito=room, ifrom=ifrom)
return await iq.send(**iqkwargs)
async def set_room_config(self, room: JID, config, *,
ifrom: Optional[JID] = None, **iqkwargs) -> Iq:
"""Send a room config form"""
query = MUCOwnerQuery()
config['type'] = 'submit'
query.append(config)
iq = self.xmpp.make_iq_set(query, ito=room, ifrom=ifrom)
return await iq.send(**iqkwargs)
async def get_affiliation_list(self, room: JID, affiliation: str, *,
ifrom: Optional[JID] = None, **iqkwargs) -> List[JID]:
""""Get a list of JIDs with the specified affiliation"""
iq = self.xmpp.make_iq_get(stanza.NS_ADMIN, ito=room, ifrom=ifrom)
iq['mucadmin_query']['item']['affiliation'] = affiliation
result = await iq.send(**iqkwargs)
return [item['jid'] for item in result['mucadmin_query']]
async def get_roles_list(self, room: JID, role: str, *,
ifrom: Optional[JID] = None, **iqkwargs) -> List[str]: ifrom: Optional[JID] = None, **iqkwargs) -> List[str]:
""""Get a list of JIDs with the specified role""" """"Get a list of JIDs with the specified role
:param room: Room to get roles from.
:param role: The role to list.
"""
iq = self.xmpp.make_iq_get(stanza.NS_ADMIN, ito=room, ifrom=ifrom) iq = self.xmpp.make_iq_get(stanza.NS_ADMIN, ito=room, ifrom=ifrom)
iq['mucadmin_query']['item']['role'] = role iq['mucadmin_query']['item']['role'] = role
result = await iq.send(**iqkwargs) result = await iq.send(**iqkwargs)
return [item['nick'] for item in result['mucadmin_query']] return [item['nick'] for item in result['mucadmin_query']]
async def send_affiliation_list(self, room: JID, affiliations: List[Tuple[JID, str]], *, async def send_role_list(self, room: JID, roles: List[Tuple[str, MucRole]], *,
ifrom: Optional[JID] = None, **iqkwargs) -> Iq: ifrom: Optional[JID] = None, **iqkwargs):
"""Send an affiliation delta list""" """Send a role delta list.
iq = self.xmpp.make_iq_set(ito=room, ifrom=ifrom)
for jid, affiliation in affiliations:
item = MUCAdminItem()
item['jid'] = jid
item['affiliation'] = affiliation
iq['mucadmin_query'].append(item)
return await iq.send(**iqkwargs)
async def send_role_list(self, room: JID, roles: List[Tuple[str, str]], *, :param room: Room to send the roles to.
ifrom: Optional[JID] = None, **iqkwargs) -> Iq: :param roles: List of couples (nick, role) to set.
"""Send a role delta list""" """
iq = self.xmpp.make_iq_set(ito=room, ifrom=ifrom) iq = self.xmpp.make_iq_set(ito=room, ifrom=ifrom)
for nick, affiliation in roles: for nick, affiliation in roles:
item = MUCAdminItem() item = MUCAdminItem()
item['nick'] = nick item['nick'] = nick
item['affiliation'] = affiliation item['affiliation'] = affiliation
iq['mucadmin_query'].append(item) iq['mucadmin_query'].append(item)
return await iq.send(**iqkwargs) await iq.send(**iqkwargs)
def invite(self, room: JID, jid: JID, reason: str = '', *,
mfrom: Optional[JID] = None):
""" Invite a jid to a room (mediated invitation).
:param room: Room to invite the user in.
:param jid: JID of the user to invite.
:param reason: Reason for inviting the user.
"""
msg = self.xmpp.make_message(room, mfrom=mfrom)
msg['muc']['invite']['to'] = jid
if reason:
msg['muc']['invite']['reason'] = reason
self.xmpp.send(msg)
def invite_server(self, room: JID, jid: JID,
invite_from: JID, reason: str = ''):
"""Send a mediated invite to a user, as a MUC service.
.. versionadded:: 1.8.0
:param room: Room to invite the user in.
:param jid: JID of the user to invite.
:param invite_from: JID of the user to send the invitation from.
:param reason: Reason for inviting the user.
"""
if not self.xmpp.is_component:
raise ValueError("Cannot use this method as a client.")
msg = self.xmpp.make_message(jid, mfrom=room)
msg['muc']['invite']['from'] = invite_from
if reason:
msg['muc']['invite']['reason'] = reason
msg.send()
def decline(self, room: JID, jid: JID, reason: str = '', *,
mfrom: Optional[JID] = None):
"""Decline a mediated invitation.
:param room: Room the invitation came from.
:param jid: JID of the user who sent the invitation.
:param reason: Reason for declining.
"""
msg = self.xmpp.make_message(room, mfrom=mfrom)
msg['muc']['decline']['to'] = jid
if reason:
msg['muc']['decline']['reason'] = reason
self.xmpp.send(msg)
def jid_in_room(self, room: JID, jid: JID) -> bool:
"""Check if a JID is present in a room.
:param room: Room to check.
:param jid: JID to check.
"""
for nick in self.rooms[room]:
entry = self.rooms[room][nick]
if not entry.get('jid'):
continue
if entry is not None and entry['jid'].full == jid:
return True
return False
def get_nick(self, room: JID, jid: JID) -> Optional[str]:
"""Get the nickname of a specific JID in a room.
:param room: Room to inspect.
:param jid: JID whose nick to return.
"""
for nick in self.rooms[room]:
entry = self.rooms[room][nick]
if not entry.get('jid'):
continue
if entry is not None and entry['jid'].full == jid:
return nick
return None
def get_joined_rooms(self) -> List[JID]: def get_joined_rooms(self) -> List[JID]:
return self.rooms.keys() """Get the list of rooms we sent a join presence to
and did not explicitly leave.
"""
return list(self.rooms.keys())
def get_our_jid_in_room(self, room_jid: JID) -> str: def get_our_jid_in_room(self, room_jid: JID) -> str:
""" Return the jid we're using in a room. """ Return the jid we're using in a room.
""" """
return "%s/%s" % (room_jid, self.our_nicks[room_jid]) return "%s/%s" % (room_jid, self.our_nicks[room_jid])
def get_jid_property(self, room, nick, jid_property): def get_jid_property(self, room: JID, nick: str,
jid_property: MucRoomItemKeys) -> Any:
""" Get the property of a nick in a room, such as its 'jid' or 'affiliation' """ Get the property of a nick in a room, such as its 'jid' or 'affiliation'
If not found, return None. If not found, return None.
:param room: Get the property for this room.
:param nick: Which nickname information to get.
:param jid_property: Property to fetch.
""" """
if room in self.rooms and nick in self.rooms[room] and jid_property in self.rooms[room][nick]: if room in self.rooms and nick in self.rooms[room] and jid_property in self.rooms[room][nick]:
return self.rooms[room][nick][jid_property] return self.rooms[room][nick][jid_property]
@ -505,10 +646,12 @@ class XEP_0045(BasePlugin):
def get_roster(self, room: JID) -> List[str]: def get_roster(self, room: JID) -> List[str]:
""" Get the list of nicks in a room. """ Get the list of nicks in a room.
:param room: Room to list nicks from.
""" """
if room not in self.rooms.keys(): if room not in self.rooms.keys():
raise ValueError("Room %s is not joined" % room) raise ValueError("Room %s is not joined" % room)
return self.rooms[room].keys() return list(self.rooms[room].keys())
def get_users_by_affiliation(self, room: JID, affiliation='member', *, ifrom: Optional[JID] = None): def get_users_by_affiliation(self, room: JID, affiliation='member', *, ifrom: Optional[JID] = None):
# Preserve old API # Preserve old API

View file

@ -7,15 +7,21 @@
This file contains boilerplate to define types relevant to slixmpp. This file contains boilerplate to define types relevant to slixmpp.
""" """
from typing import Optional
try: try:
from typing import ( from typing import (
Literal, Literal,
TypedDict,
) )
except ImportError: except ImportError:
from typing_extensions import ( from typing_extensions import (
Literal, Literal,
TypedDict,
) )
from slixmpp.jid import JID
PresenceTypes = Literal[ PresenceTypes = Literal[
'error', 'probe', 'subscribe', 'subscribed', 'error', 'probe', 'subscribe', 'subscribed',
'unavailable', 'unsubscribe', 'unsubscribed', 'unavailable', 'unsubscribe', 'unsubscribed',
@ -35,3 +41,32 @@ IqTypes = Literal[
"error", "get", "set", "result", "error", "get", "set", "result",
] ]
MucRole = Literal[
'moderator', 'participant', 'visitor', 'none'
]
MucAffiliation = Literal[
'outcast', 'member', 'admin', 'owner', 'none'
]
class PresenceArgs(TypedDict, total=False):
pfrom: JID
pto: JID
pshow: PresenceShows
ptype: PresenceTypes
pstatus: str
class MucRoomItem(TypedDict, total=False):
jid: JID
role: MucRole
affiliation: MucAffiliation
show: Optional[PresenceShows]
status: str
alt_nick: str
MucRoomItemKeys = Literal[
'jid', 'role', 'affiliation', 'show', 'status', 'alt_nick',
]