qcg/service.py

   1 import logging
   2 import time
   3
   4 from django.db import transaction
   5 from django.utils.timezone import now
   6
   7 from pyqcg.service import Registry
   8 from pyqcg.utils import Credential, TimePeriod
   9
  10
  11 logger = logging.getLogger(__name__)
  12
  13
  14 @transaction.atomic
  15 def update_user_data(user, proxy):
  16     ts = time.time()
  17     from qcg.models import User, Job, Task, Allocation, NodeInfo
  18
  19     credential = Credential(proxy)
  20     registry = Registry(credential)
  21
  22     # put lock on user record (hopefully..?)
  23     user = User.objects.select_for_update().get(pk=user.pk)
  24
  25     changed_filter = {'changed': TimePeriod(after=user.last_update)}
  26
  27     ###################################
  28     # Jobs
  29     ###################################
  30     jts = time.time()
  31     jobs = registry.jobs(**changed_filter)
  32     jte = time.time()
  33
  34     jobs_count = 0
  35     for qcg_job in jobs:
  36         params = Job.qcg_map(qcg_job, user)
  37         job_id = params.pop('job_id')
  38
  39         Job.objects.update_or_create(job_id=job_id, defaults=params)
  40         jobs_count += 1
  41
  42     ###################################
  43     # Tasks
  44     ###################################
  45     tts = time.time()
  46     tasks = registry.tasks(**changed_filter)
  47     tte = time.time()
  48
  49     jobs_cache = {j.job_id: j for j in Job.objects.filter(owner=user)}
  50     task_count = 0
  51     for qcg_task in tasks:
  52         params = Task.qcg_map(qcg_task, jobs_cache)
  53         task_id = params.pop('task_id')
  54
  55         task, created = Task.objects.update_or_create(job__job_id=qcg_task.job_id, task_id=task_id, defaults=params)
  56
  57         if not created:
  58             task.allocations.all().delete()
  59
  60         for qcg_alloc in qcg_task.allocations:
  61             alloc = task.allocations.create(**Allocation.qcg_map(qcg_alloc))
  62
  63             for qcg_node in qcg_alloc.nodes:
  64                 alloc.nodes.create(**NodeInfo.qcg_map(qcg_node))
  65
  66         task_count += 1
  67
  68     # release user lock
  69     user.last_update = now()
  70     user.save()
  71
  72     elapsed = time.time() - ts
  73     elapsed_jobs = jte - jts
  74     elapsed_tasks = tte - tts
  75     elapsed_py = elapsed - elapsed_jobs - elapsed_tasks
  76     logger.info('(%.3f) USER = %s, JOBS = %s (%.3f), TASKS = %s (%.3f), TIME = %.3f',
  77                 elapsed, user, jobs_count, elapsed_jobs, task_count, elapsed_tasks, elapsed_py)