Update job service, repo, model structure

Author: pritam001

src/model/job_model.py

@@ -0,0 +1,15 @@
+from marshmallow import Schema, fields
+
+
+class BatchJob(Schema):
+    _id = fields.Str(required=True)
+    description = fields.Str(required=True)
+    finished = fields.Bool(required=True)
+    stats = fields.Raw(required=True)
+    jobs = fields.Raw(required=True)
+    createdTime = fields.Raw()
+    updatedTime = fields.Raw()
+
+    class Meta:
+        strict = True
+        dateformat = "iso"

src/repository/job_repository.py

@@ -0,0 +1,30 @@
+from src.app import server
+from src.common import Logger
+from src.model.job_model import BatchJob
+from src.util import (
+    get_initial_create_time_dict,
+    get_update_time_dict,
+    validate_and_dump_in_schema,
+)
+
+log = Logger()
+
+
+def create_one(insert_args):
+    insert_data = {**insert_args, **get_initial_create_time_dict()}
+    log.info(f"job_repository : create_one : insert_id = {insert_data['_id']}")
+    batch_job_schema = BatchJob()
+    obj = validate_and_dump_in_schema(insert_data, batch_job_schema)
+    return server.get_mongo_db().AtxDataBatchJobs.insert_one(obj)
+
+
+def find_one(batch_job_id):
+    return server.get_mongo_db().AtxDataBatchJobs.find_one({"_id": batch_job_id})
+
+
+def update_one(batch_job_id, update_args):
+    update_data = {**update_args, **get_update_time_dict()}
+    log.info(f"job_repository : create_one : update_id = {update_data['_id']}")
+    return server.get_mongo_db().AtxDataBatchJobs.update_one(
+        {"_id": batch_job_id}, {"$set": update_data}
+    )

src/routes/health/service.py

@@ -5,7 +5,7 @@
 
 from src.app import server
 from src.common import Logger
-from src.routes.job.service import create_batch_job
+from src.services.job_service import create_batch_job
 from src.util.job import add_job_meta
 
 log = Logger()

src/routes/job/__init__.py

@@ -1,6 +1,6 @@
 from src.common.logger import Logger
 from src.core.namespace import Namespace
-from src.routes.job.service import (
+from src.services.job_service import (
     view_or_update_batch_job,
     clear_all_failed_jobs,
     clean_queue,

src/routes/job/service.py renamed to src/services/job_service.py

@@ -1,38 +1,35 @@
 import calendar
-import hashlib
-import uuid
 from datetime import datetime, timedelta
 
 from rq import Worker
 from rq.job import Job
 
-from src.common import Logger
 from src.app import server
-from src.util import get_initial_create_time_dict, get_update_time_dict
+from src.common import Logger
+from src.repository import job_repository
 from src.util.job import (
     update_job_array_with_meta,
     add_job_meta,
     is_batch_job_finished,
     batch_job_stats,
 )
+from src.util.sequence import fetch_unique_uuid_md5_id
 
 log = Logger()
 
 
 def create_batch_job(desc, job_array):
-    rand_unique_id = str(datetime.utcnow()) + str(uuid.uuid4())
-    seq_id = hashlib.md5(rand_unique_id.encode("utf-8")).hexdigest()
+    seq_id = fetch_unique_uuid_md5_id()
     log.info(f"create_batch_job : creating new batch job for {desc} with id {seq_id}")
 
     job_array_with_meta = update_job_array_with_meta(job_array)
-    insert_resp = server.get_mongo_db().AtxDataBatchJobs.insert_one(
+    insert_resp = job_repository.create_one(
         {
             "_id": seq_id,
             "description": desc,
             "finished": is_batch_job_finished(job_array_with_meta),
             "stats": batch_job_stats(job_array_with_meta),
             "jobs": job_array_with_meta,
-            **get_initial_create_time_dict(),
         }
     )
     if insert_resp.acknowledged:
@@ -78,10 +75,6 @@ def clean_queue(queue_name):
     return {"success": True}
 
 
-def fetch_batch_job(job_id):
-    return server.get_mongo_db().AtxDataBatchJobs.find_one({"_id": job_id})
-
-
 def get_all_queue_stats():
     all_stats = {}
     jq = server.get_job_queue()
@@ -128,7 +121,7 @@ def kill_all_zombie_workers():
 
 
 def view_or_update_batch_job(batch_job_id):
-    batch_job = fetch_batch_job(batch_job_id)
+    batch_job = job_repository.find_one(batch_job_id)
     if batch_job is None:
         return {
             "success": False,
@@ -152,23 +145,20 @@ def view_or_update_batch_job(batch_job_id):
             kwargs={"job_id": batch_job_id},
         )
 
-    update_resp = server.get_mongo_db().AtxDataBatchJobs.update_one(
-        {"_id": batch_job_id},
+    update_resp = job_repository.update_one(
+        batch_job_id,
         {
-            "$set": {
-                "finished": job_finished,
-                "stats": batch_job_stats(job_array_with_meta),
-                "jobs": job_array_with_meta,
-                **get_update_time_dict(),
-            }
+            "finished": job_finished,
+            "stats": batch_job_stats(job_array_with_meta),
+            "jobs": job_array_with_meta,
         },
     )
 
     if update_resp.acknowledged:
         server.get_job_queue().enqueue_job(
             poll_batch_job, priority="low", args=(tuple([batch_job_id]))
         )
-        updated_batch_job = fetch_batch_job(batch_job_id)
+        updated_batch_job = job_repository.find_one(batch_job_id)
         return {"success": True, "body": updated_batch_job}
     else:
         log.error(
@@ -193,7 +183,7 @@ def post_batch_job(batch_job_id):
 
 
 def restart_batch_job(batch_job_id):
-    batch_job = fetch_batch_job(batch_job_id)
+    batch_job = job_repository.find_one(batch_job_id)
     redis_conn = server.get_redis().get_redis_conn()
     if batch_job is None:
         return {
@@ -207,9 +197,7 @@ def restart_batch_job(batch_job_id):
             log.info(f"For batch job {batch_job_id}, requeue job {job_meta['_id']}")
             job = Job.fetch(job_meta["_id"], redis_conn)
             job.requeue()
-    server.get_mongo_db().AtxDataBatchJobs.update_one(
-        {"_id": batch_job_id}, {"$set": {"finished": False, **get_update_time_dict()}}
-    )
+    job_repository.update_one(batch_job_id, {"finished": False})
     server.get_job_queue().enqueue_job(
         poll_batch_job, priority="low", args=(tuple([batch_job_id]))
     )

src/util/__init__.py

@@ -26,9 +26,18 @@ def parse_request_using_schema(request, schema):
     func_name = inspect.stack()[1][3]
     log.info(f"Request received : {func_name}")
     request_body = request.get_json()
-    validation_errors_dict = schema.validate(request_body)
-    if len(validation_errors_dict) > 0:
-        raise ValidationError(validation_errors_dict)
+    validate_schema(request_body, schema)
     data = schema.dump(request_body)
     log.info(f"{func_name} : {data}")
     return data
+
+
+def validate_and_dump_in_schema(obj, schema):
+    validate_schema(obj, schema)
+    return schema.dump(obj)
+
+
+def validate_schema(obj, schema):
+    validation_errors_dict = schema.validate(obj)
+    if len(validation_errors_dict) > 0:
+        raise ValidationError(validation_errors_dict)

src/util/sequence.py

@@ -0,0 +1,8 @@
+import hashlib
+import uuid
+from datetime import datetime
+
+
+def fetch_unique_uuid_md5_id():
+    rand_unique_id = str(datetime.utcnow()) + str(uuid.uuid4())
+    return hashlib.md5(rand_unique_id.encode("utf-8")).hexdigest()