added database queue

app/api/endpoints/ontologies.py

@@ -39,7 +39,7 @@ async def build_from_scratch():
                 urls.append(url.decode().strip())
 
     for url in urls:
-        chain(add_ontology_task.s(url), add_ontologies.s()).apply_async()
+        chain(add_ontology_task.s(url), add_ontologies.s().set(queue='neo4j_write_queue')).apply_async()
 
     return Response(status_code=status.HTTP_204_NO_CONTENT)
 

init.sh

@@ -6,6 +6,7 @@ echo "starting swobup..."
 
 echo "starting celery"
 celery -A tasks worker -l info &
+celery -A tasks worker --queues=database_queue --concurrency=1 &
 celery -A tasks flower -l info --port=1111 &
 echo "starting uvicorn"
-uvicorn --workers 10 --host 0.0.0.0 --port 8000 main:app
+uvicorn --workers 10 --host 0.0.0.0 --port 8000 main:app

tasks.py

@@ -33,4 +33,23 @@
 app.conf["s3_endpoint_url"] = os.environ.get("s3_endpoint_url")
 app.conf["s3_region"] = os.environ.get("s3_region")
 
-print("configuration: ", app.conf)
+
+# Define the queues
+app.conf.task_queues = {
+    'default': {
+        'exchange': 'default',
+        'binding_key': 'default'
+    },
+    'database_queue': {
+        'exchange': 'database_queue',
+        'binding_key': 'database_queue'
+    }
+}
+
+# Define the routes
+app.conf.task_routes = {
+    'app.tasks.add_to_database': {'queue': 'database_queue'},
+    '*': {'queue': 'default'}  # All other tasks go to the default queue
+}
+
+print("configuration: ", app.conf)