from collections import namedtuple from pyramid.events import BeforeRender, subscriber from pyramid.renderers import render from pyramid_mailer.message import Message SerializableFlashMessage = namedtuple( "SerializableFlashMessage", ["text", "more"] ) class Ordr3Event: def __init__(self): self.request = None class FlashMessage(Ordr3Event): def __init__(self, channel, text, more): super().__init__() self.channel = channel self.text = text self.more = more @classmethod def info(cls, text, more=""): return cls("info", text, more) @classmethod def warning(cls, text, more=""): return cls("warning", text, more) @classmethod def error(cls, text, more=""): return cls("error", text, more) class EmailNotification(Ordr3Event): """ base class for user notifications """ subject = None template = None def __init__(self, user, data=None): super().__init__() self.user = user self.data = data class AccountActivationEmail(EmailNotification): """ user notification for account activation """ subject = "[ordr] Your account was activated" template = "ordr3:templates/emails/activation.jinja2" class OrderStatusChangeEmail(EmailNotification): """ user notification for order status change """ subject = "[ordr] Order Status Change" template = "ordr3:templates/emails/status_change.jinja2" class PasswordResetEmail(EmailNotification): """ user notification for password reset link """ subject = "[ordr] Password Reset" template = "ordr3:templates/emails/password_reset.jinja2" @subscriber(FlashMessage) def handle_flash_message_event(event): if event.request is None: return session = event.request.session message = SerializableFlashMessage(event.text, event.more) print(message) session.flash(message, event.channel, allow_duplicate=False) @subscriber(EmailNotification) def notify_user(event): """ notify a user about an event """ body = render( event.template, {"user": event.user, "data": event.data}, event.request ) message = Message( subject=event.subject, sender=event.request.registry.settings["mail.default_sender"], recipients=[event.user.email], body=body, ) event.request.mailer.send(message) @subscriber(BeforeRender) def add_global(event): request = event["request"] if not request.is_xhr: event.rendering_val["new_user_badge"] = request.repo.count_new_users() def emit(request, event): event.request = request request.registry.notify(event) def includeme(config): """initializing Events sytem for the Pyramid app Activate this setup using ``config.include('ordr2.events')``. """ config.add_request_method(emit, "emit")