extract method (async)

swagger_server/controllers/extract_controller.py

@@ -6,6 +6,7 @@ import json
 from pathlib import Path
 from swagger_server.models.izlusci_async_body import IzlusciAsyncBody  # noqa: E501
 from swagger_server.models.izlusci_sync_body import IzlusciSyncBody  # noqa: E501
+from swagger_server.requets_db.models.vrsta import JobManager
 from swagger_server.utils import cl_utils
 from swagger_server.util import get_random_filename, create_random_file_in_tmp_folder
 import requests
@@ -17,21 +18,6 @@ ATEapi_endpoint = "http://localhost:5000/predict"
 # ATEapi_endpoint = "http://ate-api:5000/predict"
 
 
-def get_candidates_async(body):  # noqa: E501
-    """Izlusci terminološke kandidate iz seznama besedil v conllu obliki [asinhrono, ustvari novi job]
-
-     # noqa: E501
-
-    :param body:
-    :type body: dict | bytes
-
-    :rtype: str
-    """
-    if connexion.request.is_json:
-        body = IzlusciAsyncBody.from_dict(connexion.request.get_json())  # noqa: E501
-    return 'do some magic!'
-
-
 def do_izlusci(conllus, prepovedane_besede):
     tmp_file_path = ""
     try:
@@ -67,6 +53,25 @@ def do_izlusci(conllus, prepovedane_besede):
         return str(e), 500
 
 
+def get_candidates_async(body):  # noqa: E501
+    """Izlusci terminološke kandidate iz seznama besedil v conllu obliki [asinhrono, ustvari novi job]
+
+     # noqa: E501
+
+    :param body:
+    :type body: dict | bytes
+
+    :rtype: str
+    """
+    if connexion.request.is_json:
+        body = IzlusciAsyncBody.from_dict(connexion.request.get_json())  # noqa: E501
+    job, is_old_job = JobManager.create_job(4, json.dumps(body.to_dict()))
+    if job is None:
+        return "Something went wrong", 500
+    ret = {'check_job_url': f'{connexion.request.url_root}/job/{job.id}'}
+    return ret, 200
+
+
 def get_candidates_sync(body):  # noqa: E501
     """Izlusci terminološke kandidate iz seznama besedil v conllu obliki [sihrono, rezultat v sami zahtevi]
 

swagger_server/controllers/jobs_controller.py

@@ -1,9 +1,12 @@
 import datetime
+import json
 import os.path
 
 import peewee
 import asyncio
 import concurrent.futures as cf
+
+from swagger_server.controllers.extract_controller import do_izlusci
 from swagger_server.models.job_response import JobResponse  # noqa: E501
 from swagger_server.requets_db.models.vrsta import (Job)
 from threading import Thread
@@ -14,10 +17,12 @@ import threading
 import time
 
 CLASSLA_CONCURANCE_LIMIT = 3
-DOC2TEXT_CONCURANCE_LIMIT = 4
+DOC2TEXT_CONCURANCE_LIMIT = 3
+ATEAPI_CONCURANCE_LIMIT = 2
 
 classla_sem = threading.Semaphore(CLASSLA_CONCURANCE_LIMIT)
 doc2text_sem = threading.Semaphore(DOC2TEXT_CONCURANCE_LIMIT)
+ateapi_sem = threading.Semaphore(ATEAPI_CONCURANCE_LIMIT)
 
 
 def delete_job(job_id):  # noqa: E501
@@ -69,9 +74,26 @@ def clear_up_unfinished_jobs():
 
 
 async def try_do_jobs():
-    with cf.ThreadPoolExecutor(max_workers=2) as ex:
+    with cf.ThreadPoolExecutor(max_workers=3) as ex:
         ex.submit(try_do_jobs_classla)
         ex.submit(try_do_jobs_doc2text)
+        ex.submit(try_do_jobs_ateapi)
+
+
+### Job looping
+def try_do_jobs_ateapi():
+    while True:
+        try:
+            if ateapi_sem._value > 0:
+                unfinished_jobs = Job.select() \
+                    .where(Job.finished_on.is_null(), Job.started_on.is_null(), Job.job_type == 4) \
+                    .limit(ateapi_sem._value)
+                with cf.ThreadPoolExecutor(max_workers=ATEAPI_CONCURANCE_LIMIT) as ex:
+                    [ex.submit(execute_ateapi_job, job) for job in unfinished_jobs]
+        except Exception as e:
+            print(f"Exception in try_do_jobs_ateapi")
+        finally:
+            time.sleep(3)
 
 
 ### Job looping
@@ -180,6 +202,20 @@ def execute_classla_job(job: Job):
         classla_sem.release()
 
 
+def execute_ateapi_job(job: Job):
+    try:
+        ateapi_sem.acquire()
+        job.started_on = datetime.datetime.utcnow()
+        job.save()
+        info = json.loads(job.job_input)
+        ret_json, _ = do_izlusci(info['conllus'], info['prepovedane_besede'])
+        job.job_output = ret_json
+        job.finished_on = datetime.datetime.utcnow()
+        job.save()
+    finally:
+        ateapi_sem.release()
+
+
 clear_up_unfinished_jobs()
 loop = asyncio.get_event_loop()
 

swagger_server/requets_db/models/vrsta.py

@@ -43,7 +43,7 @@ class Job(BaseModel):
     input_file = TextField(index=True, null=True)
 
 
-db.drop_tables([Job])  # TODO: After pushing this, comment it and push again
+# db.drop_tables([Job])  # TODO: After pushing this, comment it and push again
 db.create_tables([Job])
 
 
@@ -60,7 +60,7 @@ class JobManager:
         :return: Job object, Did already exist boolean
         """
         try:
-            if job_type == 2:
+            if job_type in [2, 4]:
                 job, is_new = Job.get_or_create(job_type=job_type, job_input=job_input, input_size=len(job_input))
             elif job_type in [1, 3, 12, 32]:
                 tmp_file = ""