Refactor settings

eb_sqs_worker/app_settings.py

@@ -0,0 +1,97 @@
+import warnings
+
+from django.conf import settings
+from django.core.exceptions import ImproperlyConfigured
+from django.utils.functional import cached_property
+
+
+class AppSettings:
+    def __init__(self):
+        self._queues = {}
+
+    @cached_property
+    def AWS_EB_DEFAULT_REGION(self):
+        return self._get_setting('AWS_EB_DEFAULT_REGION')
+
+    @cached_property
+    def AWS_ACCESS_KEY_ID(self):
+        return self._get_setting('AWS_ACCESS_KEY_ID')
+
+    @cached_property
+    def AWS_SECRET_ACCESS_KEY(self):
+        return self._get_setting('AWS_SECRET_ACCESS_KEY')
+
+    @cached_property
+    def AWS_EB_DEFAULT_QUEUE_NAME(self):
+        return self._get_setting('AWS_EB_DEFAULT_QUEUE_NAME',
+                                 raise_exception=False,
+                                 default='django-eb-sqs-worker')
+
+    @cached_property
+    def AWS_EB_HANDLE_SQS_TASKS(self):
+        return self._get_setting('AWS_EB_HANDLE_SQS_TASKS',
+                                 raise_exception=False,
+                                 default=False)
+
+    @cached_property
+    def AWS_EB_RUN_TASKS_LOCALLY(self):
+        return self._get_setting('AWS_EB_RUN_TASKS_LOCALLY',
+                                 raise_exception=False,
+                                 default=False)
+
+    @cached_property
+    def AWS_EB_ALERT_WHEN_EXECUTES_LONGER_THAN_SECONDS(self):
+        return self._get_setting('AWS_EB_ALERT_WHEN_EXECUTES_LONGER_THAN_SECONDS',
+                                 raise_exception=False)
+
+    @cached_property
+    def enabled_tasks(self):
+        enabled_tasks = self._get_setting('AWS_EB_ENABLED_TASKS',
+                                          raise_exception=False,
+                                          default={})
+        if not isinstance(enabled_tasks, dict):
+            raise ImproperlyConfigured(f"settings.EB_SQS[AWS_EB_ENABLED_TASKS]"
+                                       f" must be a dict, not {type(enabled_tasks)}")
+
+        return enabled_tasks
+
+    def get_queue_by_name(self, queue_name):
+        if queue_name not in self._queues:
+            from .sqs import sqs  # Avoid circular import
+            self._queues[queue_name] = sqs.get_queue_by_name(QueueName=queue_name)
+
+        return self._queues[queue_name]
+
+    @staticmethod
+    def _get_setting(name, raise_exception=True, default=None):
+        # We cannot cache this on the instance, because Django's override_settings used
+        # in test will no longer work.
+        try:
+            return getattr(settings, 'EB_SQS', {})[name]
+        except KeyError:
+            if hasattr(settings, name):
+                warnings.warn(f"Setting {name} should be migrated"
+                              f" to 'settings.EB_SQS' dictionary.", DeprecationWarning)
+                return getattr(settings, name)
+
+
+        if raise_exception:
+            raise ImproperlyConfigured(f"settings.EB_SQS[{name}] should be set.")
+        else:
+            return default
+
+    def reconfigure(self):
+        """
+        Method for testing
+
+        Since tests modify settings at runtime, this can be used to reset values that
+        should normally only be hydrated once at application startup.
+        """
+        self._queues = {}
+        self.__dict__ = {
+            k: v for k, v in self.__dict__.items()
+            if not k.startswith('AWS_')
+        }
+        self.__dict__.pop('enabled_tasks', None)
+
+app_settings = AppSettings()

eb_sqs_worker/decorators.py

@@ -2,11 +2,10 @@
 
 from functools import wraps
 
-from django.conf import settings
 from django.core.exceptions import ImproperlyConfigured
 
 from eb_sqs_worker import sqs
-
+from .app_settings import app_settings
 
 
 def task(function=None, run_locally=None, queue_name=None, task_name=None):
@@ -34,20 +33,17 @@ def actual_decorator(f):
         print(f"eb-sqs-worker: registering task {f} with decorator under name {task_name_to_use}; "
               f"Overrides: run_locally: {run_locally}, queue_name: {queue_name}, task_name: {task_name}")
 
-        if hasattr(settings, "AWS_EB_ENABLED_TASKS"):
-            if settings.AWS_EB_ENABLED_TASKS.get(task_name_to_use):
-                raise ImproperlyConfigured(f"eb-sqs-worker error while trying to register task {task_name_to_use} through "
-                                           f"decorator: task with the same name is already registered in "
-                                           f"settings.AWS_EB_ENABLED_TASKS with "
-                                           f"value {settings.AWS_EB_ENABLED_TASKS.get(task_name_to_use)}. "
-                                           f"Consider specifying task_name in @task decorator and check that "
-                                           f"your code does not reload modules with decorator functions which "
-                                           f"may force them to register multiple times.")
-        else:
-            settings.AWS_EB_ENABLED_TASKS = {}
+        if app_settings.enabled_tasks.get(task_name_to_use):
+            raise ImproperlyConfigured(f"eb-sqs-worker error while trying to register task {task_name_to_use} through "
+                                       f"decorator: task with the same name is already registered in "
+                                       f"settings.AWS_EB_ENABLED_TASKS with "
+                                       f"value {app_settings.enabled_tasks.get(task_name_to_use)}. "
+                                       f"Consider specifying task_name in @task decorator and check that "
+                                       f"your code does not reload modules with decorator functions which "
+                                       f"may force them to register multiple times.")
 
         # register task in settings
-        settings.AWS_EB_ENABLED_TASKS[task_name_to_use] = task_function_execution_path
+        app_settings.enabled_tasks[task_name_to_use] = task_function_execution_path
 
         # prepare the returned function
 
@@ -72,4 +68,4 @@ def wrapper(**kwargs):  # task functions cannot have *args, only **kwargs
         return wrapper
     if function:
         return actual_decorator(function)
-    return actual_decorator
+    return actual_decorator

eb_sqs_worker/sqs.py

@@ -1,24 +1,21 @@
 import json
+import logging
 import uuid
 
 import boto3
-from django.conf import settings
 from django.core.exceptions import ImproperlyConfigured
+from django.utils.functional import SimpleLazyObject
 from django.utils.module_loading import import_string
-import logging
 
+from .app_settings import app_settings
 
 logger = logging.getLogger(__name__)
-try:
-    AWS_REGION = settings.AWS_EB_DEFAULT_REGION
-except AttributeError:
-    raise ImproperlyConfigured("settings.AWS_EB_DEFAULT_REGION not set, please set it to use eb_sqs_worker django app")
 
-# TODO: make it lazy so we can run tests without setting this settings?
-sqs = boto3.resource('sqs',
-                     region_name=AWS_REGION,
-                     aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
-                     aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY)
+def get_sqs():
+    return boto3.resource('sqs',
+                          region_name=app_settings.AWS_REGION,
+                          aws_access_key_id=app_settings.AWS_ACCESS_KEY_ID,
+                          aws_secret_access_key=app_settings.AWS_SECRET_ACCESS_KEY)
 
 
 def send_task(task_name, task_kwargs, run_locally=None, queue_name=None):
@@ -41,7 +38,7 @@ def send_task(task_name, task_kwargs, run_locally=None, queue_name=None):
     }
 
     if run_locally is None:
-        run_locally = getattr(settings, "AWS_EB_RUN_TASKS_LOCALLY", False)
+        run_locally = getattr(app_settings, "AWS_EB_RUN_TASKS_LOCALLY", False)
 
     if run_locally:
 
@@ -57,14 +54,14 @@ def send_task(task_name, task_kwargs, run_locally=None, queue_name=None):
 
         if queue_name is None:
             try:
-                queue_name = settings.AWS_EB_DEFAULT_QUEUE_NAME
+                queue_name = app_settings.AWS_EB_DEFAULT_QUEUE_NAME
             except AttributeError:
                 raise ImproperlyConfigured("settings.AWS_EB_DEFAULT_QUEUE_NAME must be set to send task to SQS queue")
 
             # TODO: cache queues instead of looking the up every time
         try:
             # Get the queue. This returns an SQS.Queue instance
-            queue = sqs.get_queue_by_name(QueueName=queue_name)
+            queue = app_settings.get_queue_by_name(QueueName=queue_name)
         except:
             queue = sqs.create_queue(QueueName=queue_name)
 
@@ -137,17 +134,10 @@ def run_task(self):
         }
         :return:
         """
-        if not getattr(settings, "AWS_EB_ENABLED_TASKS", None):
-            raise ImproperlyConfigured(f"settings.AWS_EB_ENABLED_TASKS not set, cannot run task {self.task_name}")
-
-        if not isinstance(settings.AWS_EB_ENABLED_TASKS, dict):
-            raise ImproperlyConfigured(f"settings.AWS_EB_ENABLED_TASKS must be a dict, "
-                                       f"not {type(settings.AWS_EB_ENABLED_TASKS)}")
-
         try:
-            task_method_path = settings.AWS_EB_ENABLED_TASKS[self.task_name]
+            task_method_path = app_settings.enabled_tasks[self.task_name]
         except KeyError:
-            raise ImproperlyConfigured(f"Task named {self.task_name} is not defined in settings.AWS_EB_ENABLED_TASKS")
+            raise ImproperlyConfigured(f"Cannot run task named {self.task_name}: not registered.")
 
         task_method = import_string(task_method_path)
 
@@ -178,3 +168,6 @@ def get_pretty_info_string(self):
         result = f"{periodic_marker}Task({self.task_name}, kwargs: {self.task_kwargs}{periodic_info})"
 
         return result
+
+
+sqs = SimpleLazyObject(get_sqs)