Adding new code for tubes (#1)

Author: pyprogrammerblog

arend/__init__.py

@@ -1,37 +0,0 @@
-from arend.queue.consumer import consumer
-from multiprocessing import Pool
-from typing import List
-
-import click
-import logging
-
-logger = logging.getLogger(__name__)
-
-
-@click.command(name="Start Processor")
-@click.option('--queues', type=List[str], help='Queue names')
-@click.option('--concurrency', type=int, default=1, help='Queue concurrency')
-@click.option('--verbose', is_flag=True, default=False, help='Verbose')
-def pool_processor(queues, concurrency, verbose):
-    click.echo("Getting the Pool...")
-    p = Pool()
-    try:
-        click.echo("Starting workers...")
-        p.map(consumer, [(queue, verbose) for queue in queues] * concurrency)
-        p.close()
-    except KeyboardInterrupt:
-        click.echo("Keyboard Exception, terminating pool...")
-        p.terminate()
-        click.echo("Pool terminated...")
-    except Exception as e:
-        click.echo(f"Exception '{e}', terminating pool...")
-        p.terminate()
-        click.echo("Pool terminated...")
-    finally:
-        click.echo("Joining processes...")
-        p.join()
-        click.echo("Processes joined...")
-
-
-if __name__ == "__main__":
-    pool_processor()

arend/async_task.py

@@ -0,0 +1,18 @@
+from .mongo import MongoBackend
+from .redis import RedisBackend
+from .sql import SqlBackend
+from arend.settings import settings
+from functools import lru_cache
+
+
+@lru_cache
+def get_queue_backend():
+    backends = {
+        "redis": RedisBackend,
+        "mongo": MongoBackend,
+        "sql": SqlBackend,
+    }
+    return backends.get(settings.broker)
+
+
+QueueBroker = get_queue_backend()

arend/backend/mongo.py

@@ -0,0 +1,25 @@
+# from arend.settings import settings
+# from pymongo import MongoClient
+# from pymongo.collection import Collection
+
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+class TasksBackend:
+    def __init__(self):
+        pass
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.close()
+
+    def close(self):
+        pass
+
+    def get_(self):
+        pass

arend/backends/__init__.py

@@ -0,0 +1,27 @@
+from arend.settings import settings
+from pymongo import MongoClient
+from pymongo.collection import Collection
+
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+class MongoBackend:
+    def __init__(self):
+        self.db: MongoClient = MongoClient(settings.mongodb_string)
+        collection = settings.mongodb_notifier_task_results
+        self.tasks_collection: Collection = self.db[collection]
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.db.close()
+
+    def find_one(self):
+        pass
+
+    def update_one(self):
+        pass

arend/backends/base.py

@@ -0,0 +1,27 @@
+from arend.settings import settings
+from pymongo import MongoClient
+from pymongo.collection import Collection
+
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+class RedisBackend:
+    def __init__(self):
+        self.db: MongoClient = MongoClient(settings.mongodb_string)
+        collection = settings.mongodb_notifier_task_results
+        self.tasks_collection: Collection = self.db[collection]
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.db.close()
+
+    def find_one(self):
+        pass
+
+    def update_one(self):
+        pass

arend/backends/mongo.py

@@ -0,0 +1,27 @@
+from arend.settings import settings
+from pymongo import MongoClient
+from pymongo.collection import Collection
+
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+class SqlBackend:
+    def __init__(self):
+        self.db: MongoClient = MongoClient(settings.mongodb_string)
+        collection = settings.mongodb_notifier_task_results
+        self.tasks_collection: Collection = self.db[collection]
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.db.close()
+
+    def find_one(self):
+        pass
+
+    def update_one(self):
+        pass

arend/backends/redis.py

@@ -0,0 +1,18 @@
+from .beanstalkd import BeanstalkdBroker
+from .redis import RedisBroker
+from .sqs import SQSBroker
+from arend.settings import settings
+from functools import lru_cache
+
+
+@lru_cache
+def get_queue_broker():
+    brokers = {
+        "beanstalk": BeanstalkdBroker,
+        "redis": RedisBroker,
+        "sqs": SQSBroker,
+    }
+    return brokers.get(settings.broker)
+
+
+QueueBroker = get_queue_broker()

arend/backends/sql.py

@@ -0,0 +1,23 @@
+import logging
+import pystalkd
+import uuid
+
+
+logger = logging.getLogger(__name__)
+
+
+class BaseBroker:
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        raise NotImplementedError
+
+    def add_to_queue(self, task_uuid: uuid.UUID):
+        raise NotImplementedError
+
+    def reserve(self, timeout: int = None) -> pystalkd.Job:
+        raise NotImplementedError
+
+    def delete(self, job):
+        raise NotImplementedError