2229

from functools import wraps
from django.utils import timezone
from django.db import transaction
from django.db.models import Q
from opal.models import Patient
from intrahospital_api import logger
from intrahospital_api.exceptions import BatchLoadError
from intrahospital_api import models

MAX_ALLOWABLE_BATCH_RUN_TIME = 3600


def any_loads_running():
    """
    Returns a boolean as to whether any loads are
    running, for use by things that synch databases
    """
    patient_loading = models.InitialPatientLoad.objects.filter(
        state=models.InitialPatientLoad.RUNNING
    ).exists()

    batch_loading = models.BatchPatientLoad.objects.filter(
        state=models.BatchPatientLoad.RUNNING
    ).exists()
    result = patient_loading or batch_loading
    logger.info("Checking loads are running {}".format(result))
    return result


def good_to_go(service_name):
    """
    Are we good to run a batch load, returns True if we should.
    runs a lot of sanity checks.
    """
    batch_loads = models.BatchPatientLoad.objects.filter(
        service_name=service_name
    )

    if not batch_loads.exists():
        return True

    last_run_running = batch_loads.filter(
        Q(stopped=None) | Q(state=models.BatchPatientLoad.RUNNING)
    ).first()

    # If the previous load is still running we don't want to start a new one.
    if last_run_running:
        time_ago = timezone.now() - last_run_running.started

        if time_ago.seconds < MAX_ALLOWABLE_BATCH_RUN_TIME:
            logger.info(
                "batch still running after {} seconds".format(
                    time_ago.seconds
                )
            )
            return False
        else:
            # The batch has been running for too long. Email us to look at it
            logger.error(
                "Previous batch load for {} still running after {} seconds".format(
                    service_name, MAX_ALLOWABLE_BATCH_RUN_TIME
                )
            )
            return False

    return True


def get_batch_start_time(service_name):
    """
    If no previous successful batch run has started.
    Use the started timestamp of the first Initial Patient Load

    If a previous batch load has run and there was an initial patient
    load that finished within that time and started before it
    use the start time of the inital patient load.

    In most cases we should be using the start time of the previous
    BatchPatientLoad

    ie batch A starts at 10:00 and finishes at 10:03
    batch B will get all data since 10:00 so that nothing is lost

    however...
    the batches exclude initial patient loads, but usually that's ok
    but we have extra cracks to paper over...

    a patient load starts during the batch A load, at 9:58. It finishes
    the db load from the upstream db at 9:59 but is still saving the data
    to our db at 10:00

    it is excluded from the batch A, so batch B goes and starts its run
    from 9:59 accordingly.

    e.g.
        1. Batch A starts.
        2. Patient Y is added, their personal load is started.
        3. Batch A finishes.
        4. Batch B starts
        5. Patient Y's load finishes.
        6. Batch B stops

        7. Batch C starts.

        What time should Batch C load from?
        It starts from the time at 2
    """
    last_successful_run = models.BatchPatientLoad.objects.filter(
        state=models.BatchPatientLoad.SUCCESS,
        service_name=service_name
    ).order_by("started").last()

    if not last_successful_run:
        return models.InitialPatientLoad.objects.order_by("started").first().started

    long_patient_load = models.InitialPatientLoad.objects.filter(
        state=models.InitialPatientLoad.SUCCESS
    ).filter(
        started__lt=last_successful_run.started
    ).filter(
        stopped__gt=last_successful_run.started
    ).order_by("started").first()

    if long_patient_load:
        return long_patient_load.started
    else:
        return last_successful_run.started


def batch_load(service_name):
    """
    A decorator that runs the function as batch load
    ie wrapping it in the batch load decorator

    A Batch load is wrapped in a model called
    the BatchPatientLoad records when it starts and stops.

    The function we calling is expected to return
    a integer which should be a count of the objects
    (be they lab tests, appointments, demographics etc)
    that have changed.

    The wrapper also handles if an error is thrown it
    notifies developers that a batch load error.
    """
    def batch_load_wrapper(fun):
        @wraps(fun)
        @transaction.atomic
        def wrap(*args, **kwargs):
            if not good_to_go(service_name):
                logger.info("batch {} not ready skipping".format(service_name))
                return
            logger.info("starting batch load for {}".format(service_name))
            batch = models.BatchPatientLoad(service_name=service_name)
            batch.start()
            try:
                count = fun(*args, **kwargs)
            except Exception as e:
                batch.failed()
                logger.error("{} batch load error {}".format(
                    service_name, str(e)
                ))
            else:
                batch.count = count
                batch.complete()

                if not isinstance(count, int):
                    raise ValueError("batch load should return an integer")
        return wrap
    return batch_load_wrapper


def get_loaded_patients():
    return Patient.objects.filter(initialpatientload__state=models.InitialPatientLoad.SUCCESS)




1	from functools import wraps	1✔
2	from django.utils import timezone	1✔
3	from django.db import transaction	1✔
4	from django.db.models import Q	1✔
5	from opal.models import Patient	1✔
6	from intrahospital_api import logger	1✔
7	from intrahospital_api.exceptions import BatchLoadError	1✔
8	from intrahospital_api import models	1✔
9
10	MAX_ALLOWABLE_BATCH_RUN_TIME = 3600	1✔
11
12
13	def any_loads_running():	1✔
14	"""
15	Returns a boolean as to whether any loads are
16	running, for use by things that synch databases
17	"""
18	patient_loading = models.InitialPatientLoad.objects.filter(	1✔
19	state=models.InitialPatientLoad.RUNNING
20	).exists()
21
22	batch_loading = models.BatchPatientLoad.objects.filter(	1✔
23	state=models.BatchPatientLoad.RUNNING
24	).exists()
25	result = patient_loading or batch_loading	1✔
26	logger.info("Checking loads are running {}".format(result))	1✔
27	return result	1✔
28
29
30	def good_to_go(service_name):	1✔
31	"""
32	Are we good to run a batch load, returns True if we should.
33	runs a lot of sanity checks.
34	"""
35	batch_loads = models.BatchPatientLoad.objects.filter(	1✔
36	service_name=service_name
37	)
38
39	if not batch_loads.exists():	1✔
40	return True	1✔
41
42	last_run_running = batch_loads.filter(	1✔
43	Q(stopped=None) \| Q(state=models.BatchPatientLoad.RUNNING)
44	).first()
45
46	# If the previous load is still running we don't want to start a new one.
47	if last_run_running:	1✔
48	time_ago = timezone.now() - last_run_running.started	1✔
49
50	if time_ago.seconds < MAX_ALLOWABLE_BATCH_RUN_TIME:	1✔
51	logger.info(	×
52	"batch still running after {} seconds".format(
53	time_ago.seconds
54	)
55	)
56	return False	×
57	else:
58	# The batch has been running for too long. Email us to look at it
59	logger.error(	1✔
60	"Previous batch load for {} still running after {} seconds".format(
61	service_name, MAX_ALLOWABLE_BATCH_RUN_TIME
62	)
63	)
64	return False	1✔
65
66	return True	×
67
68
69	def get_batch_start_time(service_name):	1✔
70	"""
71	If no previous successful batch run has started.
72	Use the started timestamp of the first Initial Patient Load
73
74	If a previous batch load has run and there was an initial patient
75	load that finished within that time and started before it
76	use the start time of the inital patient load.
77
78	In most cases we should be using the start time of the previous
79	BatchPatientLoad
80
81	ie batch A starts at 10:00 and finishes at 10:03
82	batch B will get all data since 10:00 so that nothing is lost
83
84	however...
85	the batches exclude initial patient loads, but usually that's ok
86	but we have extra cracks to paper over...
87
88	a patient load starts during the batch A load, at 9:58. It finishes
89	the db load from the upstream db at 9:59 but is still saving the data
90	to our db at 10:00
91
92	it is excluded from the batch A, so batch B goes and starts its run
93	from 9:59 accordingly.
94
95	e.g.
96	1. Batch A starts.
97	2. Patient Y is added, their personal load is started.
98	3. Batch A finishes.
99	4. Batch B starts
100	5. Patient Y's load finishes.
101	6. Batch B stops
102
103	7. Batch C starts.
104
105	What time should Batch C load from?
106	It starts from the time at 2
107	"""
108	last_successful_run = models.BatchPatientLoad.objects.filter(	1✔
109	state=models.BatchPatientLoad.SUCCESS,
110	service_name=service_name
111	).order_by("started").last()
112
113	if not last_successful_run:	1✔
114	return models.InitialPatientLoad.objects.order_by("started").first().started	1✔
115
116	long_patient_load = models.InitialPatientLoad.objects.filter(	1✔
117	state=models.InitialPatientLoad.SUCCESS
118	).filter(
119	started__lt=last_successful_run.started
120	).filter(
121	stopped__gt=last_successful_run.started
122	).order_by("started").first()
123
124	if long_patient_load:	1✔
125	return long_patient_load.started	1✔
126	else:
127	return last_successful_run.started	1✔
128
129
130	def batch_load(service_name):	1✔
131	"""
132	A decorator that runs the function as batch load
133	ie wrapping it in the batch load decorator
134
135	A Batch load is wrapped in a model called
136	the BatchPatientLoad records when it starts and stops.
137
138	The function we calling is expected to return
139	a integer which should be a count of the objects
140	(be they lab tests, appointments, demographics etc)
141	that have changed.
142
143	The wrapper also handles if an error is thrown it
144	notifies developers that a batch load error.
145	"""
146	def batch_load_wrapper(fun):	1✔
147	@wraps(fun)	1✔
148	@transaction.atomic	1✔
149	def wrap(args, *kwargs):
150	if not good_to_go(service_name):	1✔
151	logger.info("batch {} not ready skipping".format(service_name))	1✔
152	return	1✔
153	logger.info("starting batch load for {}".format(service_name))	1✔
154	batch = models.BatchPatientLoad(service_name=service_name)	1✔
155	batch.start()	1✔
156	try:	1✔
157	count = fun(args, *kwargs)	1✔
158	except Exception as e:	1✔
159	batch.failed()	1✔
160	logger.error("{} batch load error {}".format(	1✔
161	service_name, str(e)
162	))
163	else:
164	batch.count = count	1✔
165	batch.complete()	1✔
166
167	if not isinstance(count, int):	1✔
168	raise ValueError("batch load should return an integer")	×
169	return wrap	1✔
170	return batch_load_wrapper	1✔
171
172
173	def get_loaded_patients():	1✔
174	return Patient.objects.filter(initialpatientload__state=models.InitialPatientLoad.SUCCESS)	1✔
175
176
177

openhealthcare / elcid-rfh / 2229

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous