common.lib.job

  1"""
  2Class that represents a job in the job queue
  3"""
  4
  5import time
  6import json
  7import math
  8from common.lib.exceptions import JobClaimedException, JobNotFoundException
  9
 10class Job:
 11	"""
 12	Job in queue
 13	"""
 14	data = {}
 15	db = None
 16
 17	is_finished = False
 18	is_claimed = False
 19
 20	def __init__(self, data, database=None):
 21		"""
 22		Instantiate Job object
 23
 24		:param dict data:  Job data, should correspond to a database record
 25		:param database:  Database handler
 26		"""
 27		self.data = data
 28		self.db = database
 29
 30		self.data["remote_id"] = str(self.data["remote_id"])
 31
 32		try:
 33			self.is_finished = "is_finished" in self.data and self.data["is_finished"]
 34			self.is_claimed = self.data["timestamp_claimed"] and self.data["timestamp_claimed"] > 0
 35		except KeyError:
 36			raise Exception
 37
 38	def get_by_ID(id, database):
 39		"""
 40		Instantiate job object by ID
 41
 42		:param int id: Job ID
 43		:param database:  Database handler
 44		:return Job: Job object
 45		"""
 46		data = database.fetchone("SELECT * FROM jobs WHERE id = %s", (id,))
 47		if not data:
 48			raise JobNotFoundException
 49
 50		return Job.get_by_data(data, database)
 51
 52	def get_by_data(data, database):
 53		"""
 54		Instantiate job object with given data
 55
 56		:param dict data:  Job data, should correspond to a database row
 57		:param database: Database handler
 58		:return Job: Job object
 59		"""
 60		return Job(data, database)
 61
 62	def get_by_remote_ID(remote_id, database, jobtype="*"):
 63		"""
 64		Instantiate job object by combination of remote ID and job type
 65
 66		This combination is guaranteed to be unique.
 67
 68		:param database: Database handler
 69		:param str jobtype: Job type
 70		:param str remote_id: Job remote ID
 71		:return Job: Job object
 72		"""
 73		if jobtype != "*":
 74			data = database.fetchone("SELECT * FROM jobs WHERE jobtype = %s AND remote_id = %s", (jobtype, remote_id))
 75		else:
 76			data = database.fetchone("SELECT * FROM jobs WHERE remote_id = %s", (remote_id,))
 77
 78		if not data:
 79			raise JobNotFoundException
 80
 81		return Job.get_by_data(data, database=database)
 82
 83	def claim(self):
 84		"""
 85		Claim a job
 86
 87		This marks it in the database so it cannot be claimed again.
 88		"""
 89		if self.data["interval"] == 0:
 90			claim_time = int(time.time())
 91		else:
 92			# the claim time should be a multiple of the interval to prevent
 93			# drift of the interval over time. this ensures that on average,
 94			# the interval remains as set
 95			claim_time = math.floor(int(time.time()) / self.data["interval"]) * self.data["interval"]
 96
 97		updated = self.db.update("jobs", data={"timestamp_claimed": claim_time, "timestamp_lastclaimed": claim_time},
 98								 where={"jobtype": self.data["jobtype"], "remote_id": self.data["remote_id"],
 99										"timestamp_claimed": 0})
100
101		if updated == 0:
102			raise JobClaimedException
103
104		self.data["timestamp_claimed"] = claim_time
105		self.data["timestamp_lastclaimed"] = claim_time
106
107		self.is_claimed = True
108
109	def finish(self, delete=False):
110		"""
111		Finish job
112
113		This deletes it from the database, or in the case of recurring jobs,
114		resets the claim flags.
115
116		:param bool delete: Whether to force deleting the job even if it is a
117							job with an interval.
118		"""
119		if self.data["interval"] == 0 or delete:
120			self.db.delete("jobs", where={"jobtype": self.data["jobtype"], "remote_id": self.data["remote_id"]})
121		else:
122			self.db.update("jobs", data={"timestamp_claimed": 0, "attempts": 0},
123						   where={"jobtype": self.data["jobtype"], "remote_id": self.data["remote_id"]})
124
125		self.is_finished = True
126
127	def release(self, delay=0, claim_after=0):
128		"""
129		Release a job so it may be claimed again
130
131		:param int delay: Delay in seconds after which job may be reclaimed.
132		:param int claim_after:  Timestamp after which job may be claimed. This
133		is overridden by `delay`.
134		"""
135		update = {"timestamp_claimed": 0, "attempts": self.data["attempts"] + 1}
136		if delay > 0:
137			update["timestamp_after"] = int(time.time()) + delay
138		elif claim_after is not None:
139			update["timestamp_after"] = claim_after
140
141		self.db.update("jobs", data=update,
142					   where={"jobtype": self.data["jobtype"], "remote_id": self.data["remote_id"]})
143		self.is_claimed = False
144
145	def is_claimable(self):
146		"""
147		Can this job be claimed?
148
149		:return bool: If the job is not claimed yet and also isn't finished.
150		"""
151		return not self.is_claimed and not self.is_finished
152
153	def get_place_in_queue(self):
154		"""
155		Get the place of this job in the queue
156
157		:return int: Place in queue
158		"""
159		query = "SELECT COUNT(*) as queue_ahead FROM jobs WHERE jobtype = %s"
160		replacements = [self.data["jobtype"]]
161		if self.data["timestamp_after"] == 0:
162			# Job can be claimed immediately
163			query += (
164				" AND (timestamp_after = 0 AND timestamp < %s OR "  # Other jobs that can be claimed immediately and were queued prior to this job being queued
165				" timestamp_after > 0 AND timestamp_after < %s) ")  # Other jobs that are waiting for a specific time, but prior to this job being queued
166			replacements += [self.data["timestamp"], self.data["timestamp"]]
167		else:
168			# Job must wait until timestamp_after
169			query += (
170				" AND (timestamp_after = 0 AND timestamp < %s OR "  # Other jobs that can be claimed immediately and were queued prior to this job
171				" timestamp_after > 0 AND timestamp_after < %s) ")  # Other jobs that are waiting, but prior to this job's start time
172			replacements += [self.data["timestamp_after"], self.data["timestamp_after"]]
173		queue_result = self.db.fetchone(query, replacements)
174		if queue_result["queue_ahead"] is None:
175			raise Exception(f"what? {queue_result}")
176
177		return queue_result["queue_ahead"]
178
179	@property
180	def details(self):
181		try:
182			details = json.loads(self.data["details"])
183			if details:
184				return details
185			else:
186				return {}
187		except (TypeError, json.JSONDecodeError):
188			return {}