summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--slixmpp/plugins/xep_0045/muc.py403
-rw-r--r--slixmpp/types.py35
2 files changed, 308 insertions, 130 deletions
diff --git a/slixmpp/plugins/xep_0045/muc.py b/slixmpp/plugins/xep_0045/muc.py
index db24addf..4507be59 100644
--- a/slixmpp/plugins/xep_0045/muc.py
+++ b/slixmpp/plugins/xep_0045/muc.py
@@ -9,6 +9,7 @@ import asyncio
import logging
from datetime import datetime
from typing import (
+ Any,
Dict,
List,
Tuple,
@@ -29,6 +30,7 @@ from slixmpp.xmlstream.matcher.stanzapath import StanzaPath
from slixmpp.xmlstream.matcher.xmlmask import MatchXMLMask
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.stanza import (
MUCInvite,
@@ -45,6 +47,13 @@ from slixmpp.plugins.xep_0045.stanza import (
MUCActor,
MUCUserItem,
)
+from slixmpp.types import (
+ MucRole,
+ MucAffiliation,
+ MucRoomItem,
+ MucRoomItemKeys,
+ PresenceArgs,
+)
log = logging.getLogger(__name__)
@@ -56,7 +65,7 @@ ROLES = ('moderator', 'participant', 'visitor', 'none')
class XEP_0045(BasePlugin):
"""
- Implements XEP-0045 Multi-User Chat
+ XEP-0045 Multi-User Chat
"""
name = 'xep_0045'
@@ -64,6 +73,9 @@ class XEP_0045(BasePlugin):
dependencies = {'xep_0030', 'xep_0004'}
stanza = stanza
+ rooms: Dict[JID, Dict[str, MucRoomItem]]
+ our_nicks: Dict[JID, str]
+
def plugin_init(self):
self.rooms = {}
self.our_nicks = {}
@@ -82,6 +94,7 @@ class XEP_0045(BasePlugin):
register_stanza_plugin(Iq, MUCAdminQuery)
register_stanza_plugin(Iq, MUCOwnerQuery)
register_stanza_plugin(MUCOwnerQuery, MUCOwnerDestroy)
+ register_stanza_plugin(MUCOwnerQuery, Form)
register_stanza_plugin(MUCAdminQuery, MUCAdminItem, iterable=True)
# Register handlers
@@ -89,7 +102,7 @@ class XEP_0045(BasePlugin):
Callback(
'MUCPresence',
StanzaPath("presence/muc"),
- self.handle_groupchat_presence,
+ self._handle_groupchat_presence,
))
# <x xmlns="http://jabber.org/protocol/muc"/> is only used in
# presence when joining on the client side, and for errors on
@@ -99,7 +112,7 @@ class XEP_0045(BasePlugin):
Callback(
'MUCPresenceJoin',
StanzaPath("presence/muc_join"),
- self.handle_groupchat_join,
+ self._handle_groupchat_join,
))
self.xmpp.register_handler(
Callback(
@@ -113,37 +126,37 @@ class XEP_0045(BasePlugin):
Callback(
'MUCError',
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(
Callback(
'MUCMessage',
MatchXMLMask("<message xmlns='%s' type='groupchat'><body/></message>" % self.xmpp.default_ns),
- self.handle_groupchat_message
+ self._handle_groupchat_message
))
self.xmpp.register_handler(
Callback(
'MUCSubject',
MatchXMLMask("<message xmlns='%s' type='groupchat'><subject/></message>" % self.xmpp.default_ns),
- self.handle_groupchat_subject
+ self._handle_groupchat_subject
))
self.xmpp.register_handler(
Callback(
'MUCConfig',
StanzaPath('message/muc/status'),
- self.handle_config_change
+ self._handle_config_change
))
self.xmpp.register_handler(
Callback(
'MUCInvite',
StanzaPath('message/muc/invite'),
- self.handle_groupchat_invite
+ self._handle_groupchat_invite
))
self.xmpp.register_handler(
Callback(
'MUCDecline',
StanzaPath('message/muc/decline'),
- self.handle_groupchat_decline
+ self._handle_groupchat_decline
))
def plugin_end(self):
@@ -152,7 +165,7 @@ class XEP_0045(BasePlugin):
def session_bind(self, jid):
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. """
if self.xmpp.is_component:
self.xmpp.event('groupchat_invite', inv)
@@ -160,7 +173,7 @@ class XEP_0045(BasePlugin):
if inv['from'] not in self.rooms.keys():
self.xmpp.event("groupchat_invite", inv)
- def handle_groupchat_decline(self, decl):
+ def _handle_groupchat_decline(self, decl: Message):
"""Handle an invitation decline."""
if self.xmpp.is_component:
self.xmpp.event('groupchat_invite', decl)
@@ -168,12 +181,12 @@ class XEP_0045(BasePlugin):
if decl['from'] in self.room.keys():
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)."""
self.xmpp.event('groupchat_config_status', 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"""
got_offline = False
got_online = False
@@ -206,61 +219,47 @@ class XEP_0045(BasePlugin):
"""Generate MUC presence error events"""
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."""
if self.xmpp.is_component:
self.xmpp.event('groupchat_presence', pr)
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)"""
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.
"""
self.xmpp.event('groupchat_message', 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.
"""
self.xmpp.event('groupchat_message_error', 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
a change of subject (or announcing it when joining the room)
"""
# See poezio#3452. A message containing subject _and_ (body or thread)
# is not a subject change.
if msg['body'] or msg['thread']:
- return None
+ return
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, *,
password: Optional[str] = None,
maxchars: Optional[int] = None,
maxstanzas: Optional[int] = None,
seconds: Optional[int] = None,
since: Optional[datetime] = None,
- presence_options: Optional[Dict[str, str]] = None,
+ presence_options: Optional[PresenceArgs] = None,
timeout: Optional[int] = None) -> Presence:
"""
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
that order.
+ .. versionadded:: 1.8.0
+
:param password: The optional room password.
:param maxchars: Max number of characters 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
stanza.send()
- future = asyncio.Future()
+ future: asyncio.Future = asyncio.Future()
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)
with context1, context2:
@@ -321,35 +322,118 @@ class XEP_0045(BasePlugin):
return pres
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.
+
+ .. deprecated:: 1.8.0
+
+ :meth:`join_muc_wait` will replace this old API starting from version
+ 1.9.0.
+
"""
- stanza = self.xmpp.make_presence(
- pto="%s/%s" % (room, nick), pstatus=pstatus,
- pshow=pshow, pfrom=pfrom
+ presence_options = PresenceArgs(
+ pshow=pshow,
+ pstatus=pstatus,
+ pfrom=pfrom,
)
- stanza.enable('muc_join')
- if password:
- stanza['muc_join']['password'] = password
+ maxchars, maxstanzas = None, None
if maxhistory:
if maxhistory == "0":
- stanza['muc_join']['history']['maxchars'] = '0'
+ maxchars = 9
else:
- stanza['muc_join']['history']['maxstanzas'] = str(maxhistory)
- self.xmpp.send(stanza)
- self.rooms[room] = {}
- self.our_nicks[room] = nick
+ maxstanzas = int(maxhistory)
+ return asyncio.ensure_future(
+ self.join_muc_wait(
+ 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):
- """Set a room’s subject."""
+ """Set a room’s subject.
+
+ :param room: JID of the room.
+ :param subject: Room subject to set.
+ """
msg = self.xmpp.make_message(room, mfrom=mfrom)
msg['type'] = 'groupchat'
msg['subject'] = subject
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):
+ """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."""
+ """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.enable('mucowner_query')
iq['mucowner_query'].enable('destroy')
@@ -359,10 +443,17 @@ class XEP_0045(BasePlugin):
iq['mucowner_query']['destroy']['reason'] = reason
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 = '',
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:
raise ValueError('%s is not a valid affiliation' % affiliation)
if not any((jid, nick)):
@@ -377,12 +468,45 @@ class XEP_0045(BasePlugin):
iq['mucadmin_query']['item']['reason'] = reason
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):
""" Change role property of a nick in a room.
Typically, roles are temporary (they last only as long as you are in the
room), whereas affiliations are permanent (they last across groupchat
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:
raise ValueError("Role %s does not exist" % role)
@@ -393,110 +517,127 @@ class XEP_0045(BasePlugin):
iq['mucadmin_query']['item']['reason'] = reason
await iq.send(**iqkwargs)
+ async def get_roles_list(self, room: JID, role: MucRole, *,
+ ifrom: Optional[JID] = None, **iqkwargs) -> List[str]:
+ """"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['mucadmin_query']['item']['role'] = role
+ result = await iq.send(**iqkwargs)
+ return [item['nick'] for item in result['mucadmin_query']]
+
+ async def send_role_list(self, room: JID, roles: List[Tuple[str, MucRole]], *,
+ ifrom: Optional[JID] = None, **iqkwargs):
+ """Send a role delta list.
+
+ :param room: Room to send the roles to.
+ :param roles: List of couples (nick, role) to set.
+ """
+ iq = self.xmpp.make_iq_set(ito=room, ifrom=ifrom)
+ for nick, affiliation in roles:
+ item = MUCAdminItem()
+ item['nick'] = nick
+ item['affiliation'] = affiliation
+ iq['mucadmin_query'].append(item)
+ await iq.send(**iqkwargs)
+
def invite(self, room: JID, jid: JID, reason: str = '', *,
mfrom: Optional[JID] = None):
- """ Invite a jid to a room."""
+ """ 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."""
+ """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 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']]
+ def jid_in_room(self, room: JID, jid: JID) -> bool:
+ """Check if a JID is present in a room.
- async def get_roles_list(self, room: JID, role: str, *,
- ifrom: Optional[JID] = None, **iqkwargs) -> List[str]:
- """"Get a list of JIDs with the specified role"""
- iq = self.xmpp.make_iq_get(stanza.NS_ADMIN, ito=room, ifrom=ifrom)
- iq['mucadmin_query']['item']['role'] = role
- result = await iq.send(**iqkwargs)
- return [item['nick'] for item in result['mucadmin_query']]
+ :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
- async def send_affiliation_list(self, room: JID, affiliations: List[Tuple[JID, str]], *,
- ifrom: Optional[JID] = None, **iqkwargs) -> Iq:
- """Send an affiliation 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)
+ def get_nick(self, room: JID, jid: JID) -> Optional[str]:
+ """Get the nickname of a specific JID in a room.
- async def send_role_list(self, room: JID, roles: List[Tuple[str, str]], *,
- ifrom: Optional[JID] = None, **iqkwargs) -> Iq:
- """Send a role delta list"""
- iq = self.xmpp.make_iq_set(ito=room, ifrom=ifrom)
- for nick, affiliation in roles:
- item = MUCAdminItem()
- item['nick'] = nick
- item['affiliation'] = affiliation
- iq['mucadmin_query'].append(item)
- return await iq.send(**iqkwargs)
+ :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]:
- 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:
""" Return the jid we're using in a room.
"""
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'
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]:
return self.rooms[room][nick][jid_property]
@@ -505,10 +646,12 @@ class XEP_0045(BasePlugin):
def get_roster(self, room: JID) -> List[str]:
""" Get the list of nicks in a room.
+
+ :param room: Room to list nicks from.
"""
if room not in self.rooms.keys():
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):
# Preserve old API
diff --git a/slixmpp/types.py b/slixmpp/types.py
index 44e24a1e..c8ab640c 100644
--- a/slixmpp/types.py
+++ b/slixmpp/types.py
@@ -7,15 +7,21 @@
This file contains boilerplate to define types relevant to slixmpp.
"""
+from typing import Optional
+
try:
from typing import (
Literal,
+ TypedDict,
)
except ImportError:
from typing_extensions import (
Literal,
+ TypedDict,
)
+from slixmpp.jid import JID
+
PresenceTypes = Literal[
'error', 'probe', 'subscribe', 'subscribed',
'unavailable', 'unsubscribe', 'unsubscribed',
@@ -35,3 +41,32 @@ IqTypes = Literal[
"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',
+]