libervia-backend: libervia/backend/plugins/plugin_misc_email

comparison libervia/backend/plugins/plugin_misc_email_invitation.py @ 4270:0d7bb4df2343

Reformatted code base using black.

author	Goffi <goffi@goffi.org>
date	Wed, 19 Jun 2024 18:44:57 +0200
parents	4b842c1fb686
children

comparison

equal deleted inserted replaced

-:64a85ce8be70
+:0d7bb4df2343
 PLUGIN_INFO = {
 C.PI_NAME: "Email Invitations",
 C.PI_IMPORT_NAME: "EMAIL_INVITATION",
 C.PI_TYPE: C.PLUG_TYPE_MISC,
-C.PI_DEPENDENCIES: ['XEP-0077'],
+C.PI_DEPENDENCIES: ["XEP-0077"],
 C.PI_RECOMMENDATIONS: ["IDENTITY"],
 C.PI_MAIN: "InvitationsPlugin",
 C.PI_HANDLER: "no",
-C.PI_DESCRIPTION: _("""invitation of people without XMPP account""")
+C.PI_DESCRIPTION: _("""invitation of people without XMPP account"""),
 }
 SUFFIX_MAX = 5
 INVITEE_PROFILE_TPL = "guest@@{uuid}"
-KEY_ID = 'id'
+KEY_ID = "id"
-KEY_JID = 'jid'
+KEY_JID = "jid"
-KEY_CREATED = 'created'
+KEY_CREATED = "created"
-KEY_LAST_CONNECTION = 'last_connection'
+KEY_LAST_CONNECTION = "last_connection"
-KEY_GUEST_PROFILE = 'guest_profile'
+KEY_GUEST_PROFILE = "guest_profile"
-KEY_PASSWORD = 'password'
+KEY_PASSWORD = "password"
-KEY_EMAILS_EXTRA = 'emails_extra'
+KEY_EMAILS_EXTRA = "emails_extra"
-EXTRA_RESERVED = {KEY_ID, KEY_JID, KEY_CREATED, 'jid_', 'jid', KEY_LAST_CONNECTION,
+EXTRA_RESERVED = {
-KEY_GUEST_PROFILE, KEY_PASSWORD, KEY_EMAILS_EXTRA}
+KEY_ID,
+KEY_JID,
+KEY_CREATED,
+"jid_",
+"jid",
+KEY_LAST_CONNECTION,
+KEY_GUEST_PROFILE,
+KEY_PASSWORD,
+KEY_EMAILS_EXTRA,
+}
 DEFAULT_SUBJECT = D_("You have been invited by {host_name} to {app_name}")
-DEFAULT_BODY = D_("""Hello {name}!
+DEFAULT_BODY = D_(
+"""Hello {name}!
 You have received an invitation from {host_name} to participate to "{app_name}".
 To join, you just have to click on the following URL:
 {url}
 Please note that this URL should not be shared with anybody!
 If you want more details on {app_name}, you can check {app_url}.
 Welcome!
-""")
+"""
+)
 class InvitationsPlugin(object):
 def __init__(self, host):
 log.info(_("plugin Invitations initialization"))
 self.host = host
-self.invitations = persistent.LazyPersistentBinaryDict('invitations')
+self.invitations = persistent.LazyPersistentBinaryDict("invitations")
-host.bridge.add_method("invitation_create", ".plugin", in_sign='sasssssssssa{ss}s',
+host.bridge.add_method(
-out_sign='a{ss}',
+"invitation_create",
-method=self._create,
+".plugin",
-async_=True)
+in_sign="sasssssssssa{ss}s",
-host.bridge.add_method("invitation_get", ".plugin", in_sign='s', out_sign='a{ss}',
+out_sign="a{ss}",
-method=self.get,
+method=self._create,
-async_=True)
+async_=True,
-host.bridge.add_method("invitation_delete", ".plugin", in_sign='s', out_sign='',
+)
-method=self._delete,
+host.bridge.add_method(
-async_=True)
+"invitation_get",
-host.bridge.add_method("invitation_modify", ".plugin", in_sign='sa{ss}b',
+".plugin",
-out_sign='',
+in_sign="s",
-method=self._modify,
+out_sign="a{ss}",
-async_=True)
+method=self.get,
-host.bridge.add_method("invitation_list", ".plugin", in_sign='s',
+async_=True,
-out_sign='a{sa{ss}}',
+)
-method=self._list,
+host.bridge.add_method(
-async_=True)
+"invitation_delete",
-host.bridge.add_method("invitation_simple_create", ".plugin", in_sign='sssss',
+".plugin",
-out_sign='a{ss}',
+in_sign="s",
-method=self._simple_create,
+out_sign="",
-async_=True)
+method=self._delete,
+async_=True,
+)
+host.bridge.add_method(
+"invitation_modify",
+".plugin",
+in_sign="sa{ss}b",
+out_sign="",
+method=self._modify,
+async_=True,
+)
+host.bridge.add_method(
+"invitation_list",
+".plugin",
+in_sign="s",
+out_sign="a{sa{ss}}",
+method=self._list,
+async_=True,
+)
+host.bridge.add_method(
+"invitation_simple_create",
+".plugin",
+in_sign="sssss",
+out_sign="a{ss}",
+method=self._simple_create,
+async_=True,
+)
 def check_extra(self, extra):
 if EXTRA_RESERVED.intersection(extra):
 raise ValueError(
-_("You can't use following key(s) in extra, they are reserved: {}")
+_(
-.format(', '.join(EXTRA_RESERVED.intersection(extra))))
+"You can't use following key(s) in extra, they are reserved: {}"
+).format(", ".join(EXTRA_RESERVED.intersection(extra)))
-def _create(self, email='', emails_extra=None, jid_='', password='', name='',
+)
-host_name='', language='', url_template='', message_subject='',
-message_body='', extra=None, profile=''):
+def _create(
+self,
+email="",
+emails_extra=None,
+jid_="",
+password="",
+name="",
+host_name="",
+language="",
+url_template="",
+message_subject="",
+message_body="",
+extra=None,
+profile="",
+):
 # XXX: we don't use **kwargs here to keep arguments name for introspection with
 #      D-Bus bridge
 if emails_extra is None:
 emails_extra = []
 if extra is None:
 extra = {}
 else:
-extra = {str(k): str(v) for k,v in extra.items()}
+extra = {str(k): str(v) for k, v in extra.items()}
-kwargs = {"extra": extra,
+kwargs = {"extra": extra, KEY_EMAILS_EXTRA: [str(e) for e in emails_extra]}
-KEY_EMAILS_EXTRA: [str(e) for e in emails_extra]
-}
 # we need to be sure that values are unicode, else they won't be pickled correctly
 # with D-Bus
-for key in ("jid_", "password", "name", "host_name", "email", "language",
+for key in (
-"url_template", "message_subject", "message_body", "profile"):
+"jid_",
+"password",
+"name",
+"host_name",
+"email",
+"language",
+"url_template",
+"message_subject",
+"message_body",
+"profile",
+):
 value = locals()[key]
 if value:
 kwargs[key] = str(value)
 return defer.ensureDeferred(self.create(**kwargs))
 if invitation.get("email") == email:
 invitation[KEY_ID] = id_
 return invitation
 async def _create_account_and_profile(
-self,
+self, id_: str, kwargs: dict, extra: dict
-id_: str,
-kwargs: dict,
-extra: dict
 ) -> None:
 """Create XMPP account and Libervia profile for guest"""
 ## XMPP account creation
-password = kwargs.pop('password', None)
+password = kwargs.pop("password", None)
 if password is None:
 password = utils.generate_password()
 assert password
 # XXX: password is here saved in clear in database
 #      it is needed for invitation as the same password is used for profile
 #      and SàT need to be able to automatically open the profile with the uuid
 # FIXME: we could add an extra encryption key which would be used with the
 #        uuid when the invitee is connecting (e.g. with URL). This key would
 #        not be saved and could be used to encrypt profile password.
 extra[KEY_PASSWORD] = password
-jid_ = kwargs.pop('jid_', None)
+jid_ = kwargs.pop("jid_", None)
 if not jid_:
-domain = self.host.memory.config_get(None, 'xmpp_domain')
+domain = self.host.memory.config_get(None, "xmpp_domain")
 if not domain:
 # TODO: fallback to profile's domain
 raise ValueError(_("You need to specify xmpp_domain in sat.conf"))
-jid_ = "invitation-{uuid}@{domain}".format(uuid=shortuuid.uuid(),
+jid_ = "invitation-{uuid}@{domain}".format(
-domain=domain)
+uuid=shortuuid.uuid(), domain=domain
+)
 jid_ = jid.JID(jid_)
 extra[KEY_JID] = jid_.full()
 if jid_.user:
 # we don't register account if there is no user as anonymous login is then
 # used
 try:
-await self.host.plugins['XEP-0077'].register_new_account(jid_, password)
+await self.host.plugins["XEP-0077"].register_new_account(jid_, password)
 except error.StanzaError as e:
 prefix = jid_.user
 idx = 0
-while e.condition == 'conflict':
+while e.condition == "conflict":
 if idx >= SUFFIX_MAX:
 raise exceptions.ConflictError(_("Can't create XMPP account"))
-jid_.user = prefix + '_' + str(idx)
+jid_.user = prefix + "_" + str(idx)
-log.info(_("requested jid already exists, trying with {}".format(
+log.info(
-jid_.full())))
+_(
+"requested jid already exists, trying with {}".format(
+jid_.full()
+)
+)
+)
 try:
-await self.host.plugins['XEP-0077'].register_new_account(
+await self.host.plugins["XEP-0077"].register_new_account(
-jid_,
+jid_, password
-password
 )
 except error.StanzaError:
 idx += 1
 else:
 break
-if e.condition != 'conflict':
+if e.condition != "conflict":
 raise e
 log.info(_("account {jid_} created").format(jid_=jid_.full()))
 ## profile creation
-extra[KEY_GUEST_PROFILE] = guest_profile = INVITEE_PROFILE_TPL.format(
+extra[KEY_GUEST_PROFILE] = guest_profile = INVITEE_PROFILE_TPL.format(uuid=id_)
-uuid=id_
-)
 # profile creation should not fail as we generate unique name ourselves
 await self.host.memory.create_profile(guest_profile, password)
 await self.host.memory.start_session(password, guest_profile)
-await self.host.memory.param_set("JabberID", jid_.full(), "Connection",
+await self.host.memory.param_set(
-profile_key=guest_profile)
+"JabberID", jid_.full(), "Connection", profile_key=guest_profile
-await self.host.memory.param_set("Password", password, "Connection",
+)
-profile_key=guest_profile)
+await self.host.memory.param_set(
+"Password", password, "Connection", profile_key=guest_profile
+)
 async def create(self, **kwargs):
 r"""Create an invitation
 This will create an XMPP account and a profile, and use a UUID to retrieve them.
 @return (dict[unicode, unicode]): dictionary with:
 - UUID associated with the invitee (key: id)
 - filled extra dictionary, as saved in the databae
 """
 ## initial checks
-extra = kwargs.pop('extra', {})
+extra = kwargs.pop("extra", {})
 if set(kwargs).intersection(extra):
 raise ValueError(
 _("You can't use following key(s) in both args and extra: {}").format(
-', '.join(set(kwargs).intersection(extra))))
+", ".join(set(kwargs).intersection(extra))
+)
+)
 self.check_extra(extra)
-email = kwargs.pop('email', None)
+email = kwargs.pop("email", None)
 existing = await self.get_existing_invitation(email)
 if existing is not None:
 log.info(f"There is already an invitation for {email!r}")
 extra.update(existing)
 del extra[KEY_ID]
-emails_extra = kwargs.pop('emails_extra', [])
+emails_extra = kwargs.pop("emails_extra", [])
 if not email and emails_extra:
 raise ValueError(
-_('You need to provide a main email address before using emails_extra'))
+_("You need to provide a main email address before using emails_extra")
+)
-if (email is not None
-and not 'url_template' in kwargs
+if (
-and not 'message_body' in kwargs):
+email is not None
+and not "url_template" in kwargs
+and not "message_body" in kwargs
+):
 raise ValueError(
-_("You need to provide url_template if you use default message body"))
+_("You need to provide url_template if you use default message body")
+)
 ## uuid
 log.info(_("creating an invitation"))
 id_ = existing[KEY_ID] if existing else str(shortuuid.uuid())
 if existing is None:
 await self._create_account_and_profile(id_, kwargs, extra)
-profile = kwargs.pop('profile', None)
+profile = kwargs.pop("profile", None)
 guest_profile = extra[KEY_GUEST_PROFILE]
 jid_ = jid.JID(extra[KEY_JID])
 ## identity
-name = kwargs.pop('name', None)
+name = kwargs.pop("name", None)
 password = extra[KEY_PASSWORD]
 if name is not None:
-extra['name'] = name
+extra["name"] = name
 try:
-id_plugin = self.host.plugins['IDENTITY']
+id_plugin = self.host.plugins["IDENTITY"]
 except KeyError:
 pass
 else:
 await self.host.connect(guest_profile, password)
 guest_client = self.host.get_client(guest_profile)
-await id_plugin.set_identity(guest_client, {'nicknames': [name]})
+await id_plugin.set_identity(guest_client, {"nicknames": [name]})
 await self.host.disconnect(guest_profile)
 ## email
-language = kwargs.pop('language', None)
+language = kwargs.pop("language", None)
 if language is not None:
-extra['language'] = language.strip()
+extra["language"] = language.strip()
 if email is not None:
-extra['email'] = email
+extra["email"] = email
 data_format.iter2dict(KEY_EMAILS_EXTRA, extra)
-url_template = kwargs.pop('url_template', '')
+url_template = kwargs.pop("url_template", "")
-format_args = {
+format_args = {"uuid": id_, "app_name": C.APP_NAME, "app_url": C.APP_URL}
-'uuid': id_,
-'app_name': C.APP_NAME,
-'app_url': C.APP_URL}
 if name is None:
-format_args['name'] = email
+format_args["name"] = email
 else:
-format_args['name'] = name
+format_args["name"] = name
 if profile is None:
-format_args['profile'] = ''
+format_args["profile"] = ""
 else:
-format_args['profile'] = extra['profile'] = profile
+format_args["profile"] = extra["profile"] = profile
-host_name = kwargs.pop('host_name', None)
+host_name = kwargs.pop("host_name", None)
 if host_name is None:
-format_args['host_name'] = profile or _("somebody")
+format_args["host_name"] = profile or _("somebody")
 else:
-format_args['host_name'] = extra['host_name'] = host_name
+format_args["host_name"] = extra["host_name"] = host_name
 invite_url = url_template.format(**format_args)
-format_args['url'] = invite_url
+format_args["url"] = invite_url
 await sat_email.send_email(
 self.host.memory.config,
 [email] + emails_extra,
-(kwargs.pop('message_subject', None) or DEFAULT_SUBJECT).format(
+(kwargs.pop("message_subject", None) or DEFAULT_SUBJECT).format(
-**format_args),
+**format_args
-(kwargs.pop('message_body', None) or DEFAULT_BODY).format(**format_args),
+),
+(kwargs.pop("message_body", None) or DEFAULT_BODY).format(**format_args),
 )
 ## roster
 # we automatically add guest to host roster (if host is specified)
 try:
 client = self.host.get_client(profile)
 except Exception as e:
 log.error(f"Can't get host profile: {profile}: {e}")
 else:
-await self.host.contact_update(client, jid_, name, ['guests'])
+await self.host.contact_update(client, jid_, name, ["guests"])
 if kwargs:
 log.warning(_("Not all arguments have been consumed: {}").format(kwargs))
 ## extra data saving
 url_template = str(url_template)
 extra = data_format.deserialise(extra_s)
 d = defer.ensureDeferred(
 self.simple_create(client, invitee_email, invitee_name, url_template, extra)
 )
-d.addCallback(lambda data: {k: str(v) for k,v in data.items()})
+d.addCallback(lambda data: {k: str(v) for k, v in data.items()})
 return d
 async def simple_create(
-self, client, invitee_email, invitee_name, url_template, extra):
+self, client, invitee_email, invitee_name, url_template, extra
+):
 """Simplified method to invite somebody by email"""
 return await self.create(
 name=invitee_name,
 email=invitee_email,
 url_template=url_template,
 async def delete(self, id_):
 """Delete an invitation data and associated XMPP account"""
 log.info(f"deleting invitation {id_}")
 data = await self.get(id_)
-guest_profile = data['guest_profile']
+guest_profile = data["guest_profile"]
-password = data['password']
+password = data["password"]
 try:
 await self.host.connect(guest_profile, password)
 guest_client = self.host.get_client(guest_profile)
 # XXX: be extra careful to use guest_client and not client below, as this will
 #   delete the associated XMPP account
 log.debug("deleting XMPP account")
-await self.host.plugins['XEP-0077'].unregister(guest_client, None)
+await self.host.plugins["XEP-0077"].unregister(guest_client, None)
 except (error.StanzaError, sasl.SASLAuthError) as e:
 log.warning(
 f"Can't delete {guest_profile}'s XMPP account, maybe it as already been "
-f"deleted: {e}")
+f"deleted: {e}"
+)
 try:
 await self.host.memory.profile_delete_async(guest_profile, True)
 except Exception as e:
 log.warning(f"Can't delete guest profile {guest_profile}: {e}")
 log.debug("removing guest data")
 await self.invitations.adel(id_)
 log.info(f"{id_} invitation has been deleted")
 def _modify(self, id_, new_extra, replace):
-return self.modify(id_, {str(k): str(v) for k,v in new_extra.items()},
+return self.modify(id_, {str(k): str(v) for k, v in new_extra.items()}, replace)
-replace)
 def modify(self, id_, new_extra, replace=False):
 """Modify invitation data
 @param id_(unicode): UUID linked to an invitation
 @param replace(bool): if True replace the data
 else update them
 @raise KeyError: there is not invitation with this id_
 """
 self.check_extra(new_extra)
 def got_current_data(current_data):
 if replace:
 new_data = new_extra
 for k in EXTRA_RESERVED:
 try:
 new_data[k] = current_data[k]
 except KeyError:
 continue
 else:
 new_data = current_data
-for k,v in new_extra.items():
+for k, v in new_extra.items():
 if k in EXTRA_RESERVED:
 log.warning(_("Skipping reserved key {key}").format(key=k))
 continue
 if v:
 new_data[k] = v
 C.PROF_KEY_NONE: don't filter invitations
 @return list(unicode): invitations uids
 """
 invitations = await self.invitations.all()
 if profile != C.PROF_KEY_NONE:
-invitations = {id_:data for id_, data in invitations.items()
+invitations = {
-if data.get('profile') == profile}
+id_: data
+for id_, data in invitations.items()
+if data.get("profile") == profile
+}
 return invitations

Mercurial > libervia-backend

comparison libervia/backend/plugins/plugin_misc_email_invitation.py @ 4270:0d7bb4df2343