Convert to Python3

ahem/backends.py

@@ -1,4 +1,4 @@
-from __future__ import unicode_literals
+
 
 import logging
 

ahem/backends.py.bak

@@ -0,0 +1,98 @@
+from __future__ import unicode_literals
+
+import logging
+
+from django.core.exceptions import ObjectDoesNotExist
+from django.core.mail import send_mail
+from django.conf import settings as django_settings
+
+from ahem.models import UserBackendRegistry
+
+
+class BaseBackend(object):
+    """
+    VARIABLES
+    name - A unique string that identifies the backend
+    across the project
+    required_settings - A list of setings required to register
+    a user in the backend
+
+    METHODS
+    send_notification - Custom method for each backend that
+    sends the desired notification
+    """
+
+    required_settings = []
+
+    @classmethod
+    def register_user(cls, user, **settings):
+        required_settings = []
+        if hasattr(cls, 'required_settings'):
+            required_settings = cls.required_settings
+
+        if not set(required_settings).issubset(set(settings.keys())):
+            raise Exception("Missing backend settings.") # TODO: change to custom exception
+
+        try:
+            registry = UserBackendRegistry.objects.get(
+                user=user, backend=cls.name)
+        except ObjectDoesNotExist:
+            registry = UserBackendRegistry(
+                user=user, backend=cls.name)
+
+        registry.settings = settings
+        registry.save()
+
+        return registry
+
+    def send_notification(self, user, notification, context={}, settings={}):
+        raise NotImplementedError
+
+
+class EmailBackend(BaseBackend):
+    """
+    CONTEXT PARAMS
+    subject - An subject for the email.
+    from_email - The email the message will be sent from,
+    default is DEFAULT_FROM_EMAIL.
+    use_html - If true, the email will be sent with html content type.
+    """
+    name = 'email'
+
+    def send_notification(self, user, notification, context={}, settings={}):
+        body = notification.render_template(user, self.name, context=context)
+        subject = context.get('subject', '')
+        from_email = context.get('from_email', django_settings.DEFAULT_FROM_EMAIL)
+        recipient_list = [user.email]
+        use_html = context.get('use_html', False)
+
+        email_params = {}
+        if use_html:
+            email_params['html_message'] = body
+
+        send_mail(subject, body, from_email, recipient_list, **email_params)
+
+
+class LoggingBackend(BaseBackend):
+    """
+    CONTEXT PARAMS
+    logger - logger name
+    logging_level - the level of the logger
+    """
+    name = 'logging'
+
+    def get_logger(self, logger_name):
+        if logger_name:
+            logger = logging.getLogger(logger_name)
+        else:
+            logger = logging
+
+        return logger
+
+    def send_notification(self, user, notification, context={}, settings={}):
+        text = notification.render_template(user, self.name, context=context)
+
+        logging_level = context.get('logging_level', 'info')
+        logger = self.get_logger(context.get('logger', None))
+
+        getattr(logger, logging_level)(text)

ahem/loader.py

@@ -1,4 +1,4 @@
-from __future__ import unicode_literals
+
 
 import inspect
 from importlib import import_module
@@ -41,7 +41,7 @@ def register_notifications():
 def get_celery_beat_schedule():
     register_notifications()
     schedule = {}
-    for _, notification in notification_registry.items():
+    for _, notification in list(notification_registry.items()):
         if notification.is_periodic():
             schedule['ahem_' + notification.name] = {
                 'task': 'ahem.tasks.dispatch_to_users',

ahem/loader.py.bak

@@ -0,0 +1,52 @@
+from __future__ import unicode_literals
+
+import inspect
+from importlib import import_module
+
+from django.conf import settings
+
+
+notification_registry = {}
+
+
+def add_to_registry(notification):
+    notification_registry[notification.name] = notification
+
+
+def load_notifications(module):
+    """
+    Loads notifications from the module.
+    """
+    from ahem.notification import Notification
+
+    for name, notification in inspect.getmembers(module):
+        if (inspect.isclass(notification) and issubclass(notification, Notification)
+                and not notification is Notification):
+            add_to_registry(notification)
+
+
+def register_notifications():
+    """
+    Registers all notifications in the application.
+    """
+    for app_path in settings.INSTALLED_APPS:
+        if app_path != 'ahem':
+            try:
+                module = import_module('%s.notifications' % app_path)
+                load_notifications(module)
+            except ImportError:
+                pass
+
+
+def get_celery_beat_schedule():
+    register_notifications()
+    schedule = {}
+    for _, notification in notification_registry.items():
+        if notification.is_periodic():
+            schedule['ahem_' + notification.name] = {
+                'task': 'ahem.tasks.dispatch_to_users',
+                'schedule': notification.trigger.crontab,
+                'args': (notification.name,)
+            }
+
+    return schedule

ahem/migrations/0001_initial.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-from __future__ import unicode_literals
+
 
 from django.db import models, migrations
 import jsonfield.fields

ahem/migrations/0001_initial.py.bak

@@ -0,0 +1,45 @@
+# -*- coding: utf-8 -*-
+from __future__ import unicode_literals
+
+from django.db import models, migrations
+import jsonfield.fields
+from django.conf import settings
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='DeferredNotification',
+            fields=[
+                ('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
+                ('notification', models.CharField(max_length=255)),
+                ('context', jsonfield.fields.JSONField(default={})),
+                ('task_id', models.CharField(max_length=255)),
+                ('ran_at', models.DateTimeField(null=True, blank=True)),
+                ('created_at', models.DateTimeField(auto_now=True)),
+            ],
+        ),
+        migrations.CreateModel(
+            name='UserBackendRegistry',
+            fields=[
+                ('id', models.AutoField(verbose_name='ID', serialize=False, auto_created=True, primary_key=True)),
+                ('backend', models.CharField(max_length=255)),
+                ('settings', jsonfield.fields.JSONField(default={})),
+                ('user', models.ForeignKey(related_name='ahem_backends', to=settings.AUTH_USER_MODEL)),
+            ],
+        ),
+        migrations.AddField(
+            model_name='deferrednotification',
+            name='user_backend',
+            field=models.ForeignKey(to='ahem.UserBackendRegistry'),
+        ),
+        migrations.AlterUniqueTogether(
+            name='userbackendregistry',
+            unique_together=set([('user', 'backend')]),
+        ),
+    ]

ahem/models.py

@@ -1,4 +1,4 @@
-from __future__ import unicode_literals
+
 
 import json
 

ahem/models.py.bak

@@ -0,0 +1,42 @@
+from __future__ import unicode_literals
+
+import json
+
+from django.db import models
+from django.conf import settings
+
+from jsonfield import JSONField
+
+from ahem.loader import register_notifications
+
+
+class UserBackendRegistry(models.Model):
+    """
+    Availble backends for a given recipient
+    """
+    user = models.ForeignKey(settings.AUTH_USER_MODEL, related_name='ahem_backends')
+    backend = models.CharField(max_length=255)
+
+    settings = JSONField(default={})
+
+    class Meta:
+        unique_together = (('user','backend'),)
+
+
+class DeferredNotification(models.Model):
+    """
+    A notification that has been deferred until a future time.
+    """
+    notification = models.CharField(max_length=255)
+    user_backend = models.ForeignKey(UserBackendRegistry)
+
+    context = JSONField(default={})
+
+    task_id = models.CharField(max_length=255)
+
+    ran_at = models.DateTimeField(null=True, blank=True)
+    created_at = models.DateTimeField(auto_now=True)
+
+
+# Notification loader
+register_notifications()

ahem/notification.py

@@ -1,4 +1,4 @@
-from __future__ import unicode_literals
+
 
 from django.utils import timezone
 from django.template.loader import get_template

ahem/notification.py.bak

@@ -0,0 +1,106 @@
+from __future__ import unicode_literals
+
+from django.utils import timezone
+from django.template.loader import get_template
+from django.template import Context, Template
+from django.db.models.query import EmptyQuerySet
+from django.contrib.auth.models import AnonymousUser
+
+from ahem.tasks import dispatch_to_users
+from ahem.utils import celery_is_available
+
+
+class Notification(object):
+    """
+    Base notification class.  Notifications extend this class.
+
+    VARIABLES
+    name - A unique string that identifies the notification
+    across the project
+    backends - A list of supported backends for this notification
+    templates - A dictionary with a backend name as the key and
+    a path for a template as value. Must have a 'default'.
+    scope - the scope of the notification.
+    trigger - A trigger class that specifies when the
+    notification will be sent.
+
+    METHODS
+    get_context_data - returns a dictionary containing context variables
+    schedule - schedules tasks according to notification configuration
+    and passed arguments
+    filter_scope - can be used to perform context based filters. Returns
+    a list of users.
+    """
+
+    def get_scope(self, backend):
+        return self.scope
+
+    def get_users(self, backend, context):
+        scope = self.get_scope(backend)
+        queryset = scope.get_users_queryset(context)
+
+        if queryset == EmptyQuerySet:
+            users = [AnonymousUser()]
+        elif hasattr(self, 'filter_scope'):
+            users = self.filter_scope(queryset, context)
+        else:
+            users = queryset.all()
+
+        return users
+
+    def get_next_run_eta(self, last_run_at=None):
+        return self.trigger.next_run_eta(last_run_at)
+
+    @classmethod
+    def is_periodic(cls):
+        return cls.trigger.is_periodic
+
+    def render_template(self, user, backend_name, context={}, **kwargs):
+        template_path = self.templates.get(backend_name)
+        if not template_path:
+            template_path = self.templates.get('default')
+
+        if not template_path:
+            raise Exception("""A template for the specified backend could not be found.
+Please define a 'default' template for the notification""")
+
+        template = get_template(template_path)
+
+        return template.render(Context(context))
+
+    def get_context_data(self, user, backend_name, **kwargs):
+        kwargs['user'] = user
+        return kwargs
+
+    def get_task_eta(self, delay_timedelta, eta):
+        run_eta = None
+        if delay_timedelta is None and eta is None:
+            run_eta = self.get_next_run_eta()
+        elif delay_timedelta:
+            run_eta = timezone.now() + delay_timedelta
+        elif eta:
+            run_eta = eta
+
+        return run_eta
+
+    def get_task_backends(self, restrict_backends):
+        if restrict_backends:
+            return list(set(self.backends).intersection(set(restrict_backends)))
+        return self.backends
+
+    def schedule(self, context={}, delay_timedelta=None, eta=None, backends=None):
+        run_eta = self.get_task_eta(delay_timedelta, eta)
+        backends = self.get_task_backends(backends)
+
+        if celery_is_available():
+            dispatch_to_users.delay(
+                self.name,
+                eta=run_eta,
+                context=context,
+                backends=backends)
+        else:
+            dispatch_to_users(
+                self.name,
+                eta=run_eta,
+                context=context,
+                backends=backends)

ahem/scopes.py

@@ -1,4 +1,4 @@
-from __future__ import unicode_literals
+
 
 from django.contrib.auth import get_user_model
 from django.db.models.query import EmptyQuerySet