We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add jitter on scheduled jobs
hourly, daily long processes if all started at once can cause sudden increase in workload if you have many sites/benches. 


Adding simple jitter to scheduled time can lessen the impact of such issues. Jitter is common pattern used for solving problems with "frequency" becomes a problem. E.g. gunicorn adds jitter to avoid restarting all workers at same time, profilers add jitter to avoid amplifying some pattern of repeated work. retry/backoff implementations also use to avoid creating patterns.


Possible implementation: When importing scheduled job types add some random delays in cron. E.g. daily jobs will start in the range of  12:00-12:15 AM instead of all starting at 12:00 AM.


Cons: Some jobs are required to be executed at specific times e.g. birthday reminders. So adding negative offset can introduce bugs for them, positive offset however should be fine AFAIK. 

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `frappe/core/doctype/scheduled_job_type/scheduled_job_type.py`
Content:
```
1 # Copyright (c) 2021, Frappe Technologies and contributors
2 # License: MIT. See LICENSE
3 
4 import json
5 from datetime import datetime
6 
7 import click
8 from croniter import CroniterBadCronError, croniter
9 
10 import frappe
11 from frappe import _
12 from frappe.model.document import Document
13 from frappe.utils import get_datetime, now_datetime
14 from frappe.utils.background_jobs import enqueue, is_job_enqueued
15 
16 
17 class ScheduledJobType(Document):
18 	# begin: auto-generated types
19 	# This code is auto-generated. Do not modify anything in this block.
20 
21 	from typing import TYPE_CHECKING
22 
23 	if TYPE_CHECKING:
24 		from frappe.types import DF
25 
26 		create_log: DF.Check
27 		cron_format: DF.Data | None
28 		frequency: DF.Literal[
29 			"All",
30 			"Hourly",
31 			"Hourly Long",
32 			"Daily",
33 			"Daily Long",
34 			"Weekly",
35 			"Weekly Long",
36 			"Monthly",
37 			"Monthly Long",
38 			"Cron",
39 			"Yearly",
40 			"Annual",
41 		]
42 		last_execution: DF.Datetime | None
43 		method: DF.Data
44 		next_execution: DF.Datetime | None
45 		server_script: DF.Link | None
46 		stopped: DF.Check
47 
48 	# end: auto-generated types
49 	def autoname(self):
50 		self.name = ".".join(self.method.split(".")[-2:])
51 
52 	def validate(self):
53 		if self.frequency != "All":
54 			# force logging for all events other than continuous ones (ALL)
55 			self.create_log = 1
56 
57 		if self.frequency == "Cron":
58 			if not self.cron_format:
59 				frappe.throw(_("Cron format is required for job types with Cron frequency."))
60 			try:
61 				croniter(self.cron_format)
62 			except CroniterBadCronError:
63 				frappe.throw(
64 					_("{0} is not a valid Cron expression.").format(f"<code>{self.cron_format}</code>"),
65 					title=_("Bad Cron Expression"),
66 				)
67 
68 	def enqueue(self, force=False) -> bool:
69 		# enqueue event if last execution is done
70 		if self.is_event_due() or force:
71 			if not self.is_job_in_queue():
72 				enqueue(
73 					"frappe.core.doctype.scheduled_job_type.scheduled_job_type.run_scheduled_job",
74 					queue=self.get_queue_name(),
75 					job_type=self.method,
76 					job_id=self.rq_job_id,
77 				)
78 				return True
79 			else:
80 				frappe.logger("scheduler").error(
81 					f"Skipped queueing {self.method} because it was found in queue for {frappe.local.site}"
82 				)
83 
84 		return False
85 
86 	def is_event_due(self, current_time=None):
87 		"""Return true if event is due based on time lapsed since last execution"""
88 		# if the next scheduled event is before NOW, then its due!
89 		return self.get_next_execution() <= (current_time or now_datetime())
90 
91 	def is_job_in_queue(self) -> bool:
92 		return is_job_enqueued(self.rq_job_id)
93 
94 	@property
95 	def rq_job_id(self):
96 		"""Unique ID created to deduplicate jobs with single RQ call."""
97 		return f"scheduled_job::{self.method}"
98 
99 	@property
100 	def next_execution(self):
101 		return self.get_next_execution()
102 
103 	def get_next_execution(self):
104 		CRON_MAP = {
105 			"Yearly": "0 0 1 1 *",
106 			"Annual": "0 0 1 1 *",
107 			"Monthly": "0 0 1 * *",
108 			"Monthly Long": "0 0 1 * *",
109 			"Weekly": "0 0 * * 0",
110 			"Weekly Long": "0 0 * * 0",
111 			"Daily": "0 0 * * *",
112 			"Daily Long": "0 0 * * *",
113 			"Hourly": "0 * * * *",
114 			"Hourly Long": "0 * * * *",
115 			"All": f"*/{(frappe.get_conf().scheduler_interval or 240) // 60} * * * *",
116 		}
117 
118 		if not self.cron_format:
119 			self.cron_format = CRON_MAP[self.frequency]
120 
121 		# If this is a cold start then last_execution will not be set.
122 		# Creation is set as fallback because if very old fallback is set job might trigger
123 		# immediately, even when it's meant to be daily.
124 		# A dynamic fallback like current time might miss the scheduler interval and job will never start.
125 		last_execution = get_datetime(self.last_execution or self.creation)
126 		return croniter(self.cron_format, last_execution).get_next(datetime)
127 
128 	def execute(self):
129 		self.scheduler_log = None
130 		try:
131 			self.log_status("Start")
132 			if self.server_script:
133 				script_name = frappe.db.get_value("Server Script", self.server_script)
134 				if script_name:
135 					frappe.get_doc("Server Script", script_name).execute_scheduled_method()
136 			else:
137 				frappe.get_attr(self.method)()
138 			frappe.db.commit()
139 			self.log_status("Complete")
140 		except Exception:
141 			frappe.db.rollback()
142 			self.log_status("Failed")
143 
144 	def log_status(self, status):
145 		# log file
146 		frappe.logger("scheduler").info(f"Scheduled Job {status}: {self.method} for {frappe.local.site}")
147 		self.update_scheduler_log(status)
148 
149 	def update_scheduler_log(self, status):
150 		if not self.create_log:
151 			# self.get_next_execution will work properly iff self.last_execution is properly set
152 			if self.frequency == "All" and status == "Start":
153 				self.db_set("last_execution", now_datetime(), update_modified=False)
154 				frappe.db.commit()
155 			return
156 		if not self.scheduler_log:
157 			self.scheduler_log = frappe.get_doc(
158 				dict(doctype="Scheduled Job Log", scheduled_job_type=self.name)
159 			).insert(ignore_permissions=True)
160 		self.scheduler_log.db_set("status", status)
161 		if status == "Failed":
162 			self.scheduler_log.db_set("details", frappe.get_traceback(with_context=True))
163 		if status == "Start":
164 			self.db_set("last_execution", now_datetime(), update_modified=False)
165 		frappe.db.commit()
166 
167 	def get_queue_name(self):
168 		return "long" if ("Long" in self.frequency) else "default"
169 
170 	def on_trash(self):
171 		frappe.db.delete("Scheduled Job Log", {"scheduled_job_type": self.name})
172 
173 
174 @frappe.whitelist()
175 def execute_event(doc: str):
176 	frappe.only_for("System Manager")
177 	doc = json.loads(doc)
178 	frappe.get_doc("Scheduled Job Type", doc.get("name")).enqueue(force=True)
179 	return doc
180 
181 
182 def run_scheduled_job(job_type: str):
183 	"""This is a wrapper function that runs a hooks.scheduler_events method"""
184 	try:
185 		frappe.get_doc("Scheduled Job Type", dict(method=job_type)).execute()
186 	except Exception:
187 		print(frappe.get_traceback())
188 
189 
190 def sync_jobs(hooks: dict | None = None):
191 	frappe.reload_doc("core", "doctype", "scheduled_job_type")
192 	scheduler_events = hooks or frappe.get_hooks("scheduler_events")
193 	all_events = insert_events(scheduler_events)
194 	clear_events(all_events)
195 
196 
197 def insert_events(scheduler_events: dict) -> list:
198 	cron_jobs, event_jobs = [], []
199 	for event_type in scheduler_events:
200 		events = scheduler_events.get(event_type)
201 		if isinstance(events, dict):
202 			cron_jobs += insert_cron_jobs(events)
203 		else:
204 			# hourly, daily etc
205 			event_jobs += insert_event_jobs(events, event_type)
206 	return cron_jobs + event_jobs
207 
208 
209 def insert_cron_jobs(events: dict) -> list:
210 	cron_jobs = []
211 	for cron_format in events:
212 		for event in events.get(cron_format):
213 			cron_jobs.append(event)
214 			insert_single_event("Cron", event, cron_format)
215 	return cron_jobs
216 
217 
218 def insert_event_jobs(events: list, event_type: str) -> list:
219 	event_jobs = []
220 	for event in events:
221 		event_jobs.append(event)
222 		frequency = event_type.replace("_", " ").title()
223 		insert_single_event(frequency, event)
224 	return event_jobs
225 
226 
227 def insert_single_event(frequency: str, event: str, cron_format: str | None = None):
228 	cron_expr = {"cron_format": cron_format} if cron_format else {}
229 
230 	try:
231 		frappe.get_attr(event)
232 	except Exception as e:
233 		click.secho(f"{event} is not a valid method: {e}", fg="yellow")
234 
235 	doc = frappe.get_doc(
236 		{
237 			"doctype": "Scheduled Job Type",
238 			"method": event,
239 			"cron_format": cron_format,
240 			"frequency": frequency,
241 		}
242 	)
243 
244 	if not frappe.db.exists("Scheduled Job Type", {"method": event, "frequency": frequency, **cron_expr}):
245 		savepoint = "scheduled_job_type_creation"
246 		try:
247 			frappe.db.savepoint(savepoint)
248 			doc.insert()
249 		except frappe.DuplicateEntryError:
250 			frappe.db.rollback(save_point=savepoint)
251 			doc.delete()
252 			doc.insert()
253 
254 
255 def clear_events(all_events: list):
256 	for event in frappe.get_all("Scheduled Job Type", fields=["name", "method", "server_script"]):
257 		is_server_script = event.server_script
258 		is_defined_in_hooks = event.method in all_events
259 
260 		if not (is_defined_in_hooks or is_server_script):
261 			frappe.delete_doc("Scheduled Job Type", event.name)
262
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/frappe/core/doctype/scheduled_job_type/scheduled_job_type.py b/frappe/core/doctype/scheduled_job_type/scheduled_job_type.py
--- a/frappe/core/doctype/scheduled_job_type/scheduled_job_type.py
+++ b/frappe/core/doctype/scheduled_job_type/scheduled_job_type.py
@@ -2,7 +2,8 @@
 # License: MIT. See LICENSE
 
 import json
-from datetime import datetime
+from datetime import datetime, timedelta
+from random import randint
 
 import click
 from croniter import CroniterBadCronError, croniter
@@ -123,7 +124,12 @@
 		# immediately, even when it's meant to be daily.
 		# A dynamic fallback like current time might miss the scheduler interval and job will never start.
 		last_execution = get_datetime(self.last_execution or self.creation)
-		return croniter(self.cron_format, last_execution).get_next(datetime)
+		next_execution = croniter(self.cron_format, last_execution).get_next(datetime)
+
+		jitter = 0
+		if self.frequency in ("Hourly Long", "Daily Long"):
+			jitter = randint(1, 600)
+		return next_execution + timedelta(seconds=jitter)
 
 	def execute(self):
 		self.scheduler_log = None

{"golden_diff": "diff --git a/frappe/core/doctype/scheduled_job_type/scheduled_job_type.py b/frappe/core/doctype/scheduled_job_type/scheduled_job_type.py\n--- a/frappe/core/doctype/scheduled_job_type/scheduled_job_type.py\n+++ b/frappe/core/doctype/scheduled_job_type/scheduled_job_type.py\n@@ -2,7 +2,8 @@\n # License: MIT. See LICENSE\n \n import json\n-from datetime import datetime\n+from datetime import datetime, timedelta\n+from random import randint\n \n import click\n from croniter import CroniterBadCronError, croniter\n@@ -123,7 +124,12 @@\n \t\t# immediately, even when it's meant to be daily.\n \t\t# A dynamic fallback like current time might miss the scheduler interval and job will never start.\n \t\tlast_execution = get_datetime(self.last_execution or self.creation)\n-\t\treturn croniter(self.cron_format, last_execution).get_next(datetime)\n+\t\tnext_execution = croniter(self.cron_format, last_execution).get_next(datetime)\n+\n+\t\tjitter = 0\n+\t\tif self.frequency in (\"Hourly Long\", \"Daily Long\"):\n+\t\t\tjitter = randint(1, 600)\n+\t\treturn next_execution + timedelta(seconds=jitter)\n \n \tdef execute(self):\n \t\tself.scheduler_log = None\n", "issue": "Add jitter on scheduled jobs\nhourly, daily long processes if all started at once can cause sudden increase in workload if you have many sites/benches. \r\n\r\n\r\nAdding simple jitter to scheduled time can lessen the impact of such issues. Jitter is common pattern used for solving problems with \"frequency\" becomes a problem. E.g. gunicorn adds jitter to avoid restarting all workers at same time, profilers add jitter to avoid amplifying some pattern of repeated work. retry/backoff implementations also use to avoid creating patterns.\r\n\r\n\r\nPossible implementation: When importing scheduled job types add some random delays in cron. E.g. daily jobs will start in the range of  12:00-12:15 AM instead of all starting at 12:00 AM.\r\n\r\n\r\nCons: Some jobs are required to be executed at specific times e.g. birthday reminders. So adding negative offset can introduce bugs for them, positive offset however should be fine AFAIK. \n", "before_files": [{"content": "# Copyright (c) 2021, Frappe Technologies and contributors\n# License: MIT. See LICENSE\n\nimport json\nfrom datetime import datetime\n\nimport click\nfrom croniter import CroniterBadCronError, croniter\n\nimport frappe\nfrom frappe import _\nfrom frappe.model.document import Document\nfrom frappe.utils import get_datetime, now_datetime\nfrom frappe.utils.background_jobs import enqueue, is_job_enqueued\n\n\nclass ScheduledJobType(Document):\n\t# begin: auto-generated types\n\t# This code is auto-generated. Do not modify anything in this block.\n\n\tfrom typing import TYPE_CHECKING\n\n\tif TYPE_CHECKING:\n\t\tfrom frappe.types import DF\n\n\t\tcreate_log: DF.Check\n\t\tcron_format: DF.Data | None\n\t\tfrequency: DF.Literal[\n\t\t\t\"All\",\n\t\t\t\"Hourly\",\n\t\t\t\"Hourly Long\",\n\t\t\t\"Daily\",\n\t\t\t\"Daily Long\",\n\t\t\t\"Weekly\",\n\t\t\t\"Weekly Long\",\n\t\t\t\"Monthly\",\n\t\t\t\"Monthly Long\",\n\t\t\t\"Cron\",\n\t\t\t\"Yearly\",\n\t\t\t\"Annual\",\n\t\t]\n\t\tlast_execution: DF.Datetime | None\n\t\tmethod: DF.Data\n\t\tnext_execution: DF.Datetime | None\n\t\tserver_script: DF.Link | None\n\t\tstopped: DF.Check\n\n\t# end: auto-generated types\n\tdef autoname(self):\n\t\tself.name = \".\".join(self.method.split(\".\")[-2:])\n\n\tdef validate(self):\n\t\tif self.frequency != \"All\":\n\t\t\t# force logging for all events other than continuous ones (ALL)\n\t\t\tself.create_log = 1\n\n\t\tif self.frequency == \"Cron\":\n\t\t\tif not self.cron_format:\n\t\t\t\tfrappe.throw(_(\"Cron format is required for job types with Cron frequency.\"))\n\t\t\ttry:\n\t\t\t\tcroniter(self.cron_format)\n\t\t\texcept CroniterBadCronError:\n\t\t\t\tfrappe.throw(\n\t\t\t\t\t_(\"{0} is not a valid Cron expression.\").format(f\"<code>{self.cron_format}</code>\"),\n\t\t\t\t\ttitle=_(\"Bad Cron Expression\"),\n\t\t\t\t)\n\n\tdef enqueue(self, force=False) -> bool:\n\t\t# enqueue event if last execution is done\n\t\tif self.is_event_due() or force:\n\t\t\tif not self.is_job_in_queue():\n\t\t\t\tenqueue(\n\t\t\t\t\t\"frappe.core.doctype.scheduled_job_type.scheduled_job_type.run_scheduled_job\",\n\t\t\t\t\tqueue=self.get_queue_name(),\n\t\t\t\t\tjob_type=self.method,\n\t\t\t\t\tjob_id=self.rq_job_id,\n\t\t\t\t)\n\t\t\t\treturn True\n\t\t\telse:\n\t\t\t\tfrappe.logger(\"scheduler\").error(\n\t\t\t\t\tf\"Skipped queueing {self.method} because it was found in queue for {frappe.local.site}\"\n\t\t\t\t)\n\n\t\treturn False\n\n\tdef is_event_due(self, current_time=None):\n\t\t\"\"\"Return true if event is due based on time lapsed since last execution\"\"\"\n\t\t# if the next scheduled event is before NOW, then its due!\n\t\treturn self.get_next_execution() <= (current_time or now_datetime())\n\n\tdef is_job_in_queue(self) -> bool:\n\t\treturn is_job_enqueued(self.rq_job_id)\n\n\t@property\n\tdef rq_job_id(self):\n\t\t\"\"\"Unique ID created to deduplicate jobs with single RQ call.\"\"\"\n\t\treturn f\"scheduled_job::{self.method}\"\n\n\t@property\n\tdef next_execution(self):\n\t\treturn self.get_next_execution()\n\n\tdef get_next_execution(self):\n\t\tCRON_MAP = {\n\t\t\t\"Yearly\": \"0 0 1 1 *\",\n\t\t\t\"Annual\": \"0 0 1 1 *\",\n\t\t\t\"Monthly\": \"0 0 1 * *\",\n\t\t\t\"Monthly Long\": \"0 0 1 * *\",\n\t\t\t\"Weekly\": \"0 0 * * 0\",\n\t\t\t\"Weekly Long\": \"0 0 * * 0\",\n\t\t\t\"Daily\": \"0 0 * * *\",\n\t\t\t\"Daily Long\": \"0 0 * * *\",\n\t\t\t\"Hourly\": \"0 * * * *\",\n\t\t\t\"Hourly Long\": \"0 * * * *\",\n\t\t\t\"All\": f\"*/{(frappe.get_conf().scheduler_interval or 240) // 60} * * * *\",\n\t\t}\n\n\t\tif not self.cron_format:\n\t\t\tself.cron_format = CRON_MAP[self.frequency]\n\n\t\t# If this is a cold start then last_execution will not be set.\n\t\t# Creation is set as fallback because if very old fallback is set job might trigger\n\t\t# immediately, even when it's meant to be daily.\n\t\t# A dynamic fallback like current time might miss the scheduler interval and job will never start.\n\t\tlast_execution = get_datetime(self.last_execution or self.creation)\n\t\treturn croniter(self.cron_format, last_execution).get_next(datetime)\n\n\tdef execute(self):\n\t\tself.scheduler_log = None\n\t\ttry:\n\t\t\tself.log_status(\"Start\")\n\t\t\tif self.server_script:\n\t\t\t\tscript_name = frappe.db.get_value(\"Server Script\", self.server_script)\n\t\t\t\tif script_name:\n\t\t\t\t\tfrappe.get_doc(\"Server Script\", script_name).execute_scheduled_method()\n\t\t\telse:\n\t\t\t\tfrappe.get_attr(self.method)()\n\t\t\tfrappe.db.commit()\n\t\t\tself.log_status(\"Complete\")\n\t\texcept Exception:\n\t\t\tfrappe.db.rollback()\n\t\t\tself.log_status(\"Failed\")\n\n\tdef log_status(self, status):\n\t\t# log file\n\t\tfrappe.logger(\"scheduler\").info(f\"Scheduled Job {status}: {self.method} for {frappe.local.site}\")\n\t\tself.update_scheduler_log(status)\n\n\tdef update_scheduler_log(self, status):\n\t\tif not self.create_log:\n\t\t\t# self.get_next_execution will work properly iff self.last_execution is properly set\n\t\t\tif self.frequency == \"All\" and status == \"Start\":\n\t\t\t\tself.db_set(\"last_execution\", now_datetime(), update_modified=False)\n\t\t\t\tfrappe.db.commit()\n\t\t\treturn\n\t\tif not self.scheduler_log:\n\t\t\tself.scheduler_log = frappe.get_doc(\n\t\t\t\tdict(doctype=\"Scheduled Job Log\", scheduled_job_type=self.name)\n\t\t\t).insert(ignore_permissions=True)\n\t\tself.scheduler_log.db_set(\"status\", status)\n\t\tif status == \"Failed\":\n\t\t\tself.scheduler_log.db_set(\"details\", frappe.get_traceback(with_context=True))\n\t\tif status == \"Start\":\n\t\t\tself.db_set(\"last_execution\", now_datetime(), update_modified=False)\n\t\tfrappe.db.commit()\n\n\tdef get_queue_name(self):\n\t\treturn \"long\" if (\"Long\" in self.frequency) else \"default\"\n\n\tdef on_trash(self):\n\t\tfrappe.db.delete(\"Scheduled Job Log\", {\"scheduled_job_type\": self.name})\n\n\n@frappe.whitelist()\ndef execute_event(doc: str):\n\tfrappe.only_for(\"System Manager\")\n\tdoc = json.loads(doc)\n\tfrappe.get_doc(\"Scheduled Job Type\", doc.get(\"name\")).enqueue(force=True)\n\treturn doc\n\n\ndef run_scheduled_job(job_type: str):\n\t\"\"\"This is a wrapper function that runs a hooks.scheduler_events method\"\"\"\n\ttry:\n\t\tfrappe.get_doc(\"Scheduled Job Type\", dict(method=job_type)).execute()\n\texcept Exception:\n\t\tprint(frappe.get_traceback())\n\n\ndef sync_jobs(hooks: dict | None = None):\n\tfrappe.reload_doc(\"core\", \"doctype\", \"scheduled_job_type\")\n\tscheduler_events = hooks or frappe.get_hooks(\"scheduler_events\")\n\tall_events = insert_events(scheduler_events)\n\tclear_events(all_events)\n\n\ndef insert_events(scheduler_events: dict) -> list:\n\tcron_jobs, event_jobs = [], []\n\tfor event_type in scheduler_events:\n\t\tevents = scheduler_events.get(event_type)\n\t\tif isinstance(events, dict):\n\t\t\tcron_jobs += insert_cron_jobs(events)\n\t\telse:\n\t\t\t# hourly, daily etc\n\t\t\tevent_jobs += insert_event_jobs(events, event_type)\n\treturn cron_jobs + event_jobs\n\n\ndef insert_cron_jobs(events: dict) -> list:\n\tcron_jobs = []\n\tfor cron_format in events:\n\t\tfor event in events.get(cron_format):\n\t\t\tcron_jobs.append(event)\n\t\t\tinsert_single_event(\"Cron\", event, cron_format)\n\treturn cron_jobs\n\n\ndef insert_event_jobs(events: list, event_type: str) -> list:\n\tevent_jobs = []\n\tfor event in events:\n\t\tevent_jobs.append(event)\n\t\tfrequency = event_type.replace(\"_\", \" \").title()\n\t\tinsert_single_event(frequency, event)\n\treturn event_jobs\n\n\ndef insert_single_event(frequency: str, event: str, cron_format: str | None = None):\n\tcron_expr = {\"cron_format\": cron_format} if cron_format else {}\n\n\ttry:\n\t\tfrappe.get_attr(event)\n\texcept Exception as e:\n\t\tclick.secho(f\"{event} is not a valid method: {e}\", fg=\"yellow\")\n\n\tdoc = frappe.get_doc(\n\t\t{\n\t\t\t\"doctype\": \"Scheduled Job Type\",\n\t\t\t\"method\": event,\n\t\t\t\"cron_format\": cron_format,\n\t\t\t\"frequency\": frequency,\n\t\t}\n\t)\n\n\tif not frappe.db.exists(\"Scheduled Job Type\", {\"method\": event, \"frequency\": frequency, **cron_expr}):\n\t\tsavepoint = \"scheduled_job_type_creation\"\n\t\ttry:\n\t\t\tfrappe.db.savepoint(savepoint)\n\t\t\tdoc.insert()\n\t\texcept frappe.DuplicateEntryError:\n\t\t\tfrappe.db.rollback(save_point=savepoint)\n\t\t\tdoc.delete()\n\t\t\tdoc.insert()\n\n\ndef clear_events(all_events: list):\n\tfor event in frappe.get_all(\"Scheduled Job Type\", fields=[\"name\", \"method\", \"server_script\"]):\n\t\tis_server_script = event.server_script\n\t\tis_defined_in_hooks = event.method in all_events\n\n\t\tif not (is_defined_in_hooks or is_server_script):\n\t\t\tfrappe.delete_doc(\"Scheduled Job Type\", event.name)\n", "path": "frappe/core/doctype/scheduled_job_type/scheduled_job_type.py"}], "after_files": [{"content": "# Copyright (c) 2021, Frappe Technologies and contributors\n# License: MIT. See LICENSE\n\nimport json\nfrom datetime import datetime, timedelta\nfrom random import randint\n\nimport click\nfrom croniter import CroniterBadCronError, croniter\n\nimport frappe\nfrom frappe import _\nfrom frappe.model.document import Document\nfrom frappe.utils import get_datetime, now_datetime\nfrom frappe.utils.background_jobs import enqueue, is_job_enqueued\n\n\nclass ScheduledJobType(Document):\n\t# begin: auto-generated types\n\t# This code is auto-generated. Do not modify anything in this block.\n\n\tfrom typing import TYPE_CHECKING\n\n\tif TYPE_CHECKING:\n\t\tfrom frappe.types import DF\n\n\t\tcreate_log: DF.Check\n\t\tcron_format: DF.Data | None\n\t\tfrequency: DF.Literal[\n\t\t\t\"All\",\n\t\t\t\"Hourly\",\n\t\t\t\"Hourly Long\",\n\t\t\t\"Daily\",\n\t\t\t\"Daily Long\",\n\t\t\t\"Weekly\",\n\t\t\t\"Weekly Long\",\n\t\t\t\"Monthly\",\n\t\t\t\"Monthly Long\",\n\t\t\t\"Cron\",\n\t\t\t\"Yearly\",\n\t\t\t\"Annual\",\n\t\t]\n\t\tlast_execution: DF.Datetime | None\n\t\tmethod: DF.Data\n\t\tnext_execution: DF.Datetime | None\n\t\tserver_script: DF.Link | None\n\t\tstopped: DF.Check\n\n\t# end: auto-generated types\n\tdef autoname(self):\n\t\tself.name = \".\".join(self.method.split(\".\")[-2:])\n\n\tdef validate(self):\n\t\tif self.frequency != \"All\":\n\t\t\t# force logging for all events other than continuous ones (ALL)\n\t\t\tself.create_log = 1\n\n\t\tif self.frequency == \"Cron\":\n\t\t\tif not self.cron_format:\n\t\t\t\tfrappe.throw(_(\"Cron format is required for job types with Cron frequency.\"))\n\t\t\ttry:\n\t\t\t\tcroniter(self.cron_format)\n\t\t\texcept CroniterBadCronError:\n\t\t\t\tfrappe.throw(\n\t\t\t\t\t_(\"{0} is not a valid Cron expression.\").format(f\"<code>{self.cron_format}</code>\"),\n\t\t\t\t\ttitle=_(\"Bad Cron Expression\"),\n\t\t\t\t)\n\n\tdef enqueue(self, force=False) -> bool:\n\t\t# enqueue event if last execution is done\n\t\tif self.is_event_due() or force:\n\t\t\tif not self.is_job_in_queue():\n\t\t\t\tenqueue(\n\t\t\t\t\t\"frappe.core.doctype.scheduled_job_type.scheduled_job_type.run_scheduled_job\",\n\t\t\t\t\tqueue=self.get_queue_name(),\n\t\t\t\t\tjob_type=self.method,\n\t\t\t\t\tjob_id=self.rq_job_id,\n\t\t\t\t)\n\t\t\t\treturn True\n\t\t\telse:\n\t\t\t\tfrappe.logger(\"scheduler\").error(\n\t\t\t\t\tf\"Skipped queueing {self.method} because it was found in queue for {frappe.local.site}\"\n\t\t\t\t)\n\n\t\treturn False\n\n\tdef is_event_due(self, current_time=None):\n\t\t\"\"\"Return true if event is due based on time lapsed since last execution\"\"\"\n\t\t# if the next scheduled event is before NOW, then its due!\n\t\treturn self.get_next_execution() <= (current_time or now_datetime())\n\n\tdef is_job_in_queue(self) -> bool:\n\t\treturn is_job_enqueued(self.rq_job_id)\n\n\t@property\n\tdef rq_job_id(self):\n\t\t\"\"\"Unique ID created to deduplicate jobs with single RQ call.\"\"\"\n\t\treturn f\"scheduled_job::{self.method}\"\n\n\t@property\n\tdef next_execution(self):\n\t\treturn self.get_next_execution()\n\n\tdef get_next_execution(self):\n\t\tCRON_MAP = {\n\t\t\t\"Yearly\": \"0 0 1 1 *\",\n\t\t\t\"Annual\": \"0 0 1 1 *\",\n\t\t\t\"Monthly\": \"0 0 1 * *\",\n\t\t\t\"Monthly Long\": \"0 0 1 * *\",\n\t\t\t\"Weekly\": \"0 0 * * 0\",\n\t\t\t\"Weekly Long\": \"0 0 * * 0\",\n\t\t\t\"Daily\": \"0 0 * * *\",\n\t\t\t\"Daily Long\": \"0 0 * * *\",\n\t\t\t\"Hourly\": \"0 * * * *\",\n\t\t\t\"Hourly Long\": \"0 * * * *\",\n\t\t\t\"All\": f\"*/{(frappe.get_conf().scheduler_interval or 240) // 60} * * * *\",\n\t\t}\n\n\t\tif not self.cron_format:\n\t\t\tself.cron_format = CRON_MAP[self.frequency]\n\n\t\t# If this is a cold start then last_execution will not be set.\n\t\t# Creation is set as fallback because if very old fallback is set job might trigger\n\t\t# immediately, even when it's meant to be daily.\n\t\t# A dynamic fallback like current time might miss the scheduler interval and job will never start.\n\t\tlast_execution = get_datetime(self.last_execution or self.creation)\n\t\tnext_execution = croniter(self.cron_format, last_execution).get_next(datetime)\n\n\t\tjitter = 0\n\t\tif self.frequency in (\"Hourly Long\", \"Daily Long\"):\n\t\t\tjitter = randint(1, 600)\n\t\treturn next_execution + timedelta(seconds=jitter)\n\n\tdef execute(self):\n\t\tself.scheduler_log = None\n\t\ttry:\n\t\t\tself.log_status(\"Start\")\n\t\t\tif self.server_script:\n\t\t\t\tscript_name = frappe.db.get_value(\"Server Script\", self.server_script)\n\t\t\t\tif script_name:\n\t\t\t\t\tfrappe.get_doc(\"Server Script\", script_name).execute_scheduled_method()\n\t\t\telse:\n\t\t\t\tfrappe.get_attr(self.method)()\n\t\t\tfrappe.db.commit()\n\t\t\tself.log_status(\"Complete\")\n\t\texcept Exception:\n\t\t\tfrappe.db.rollback()\n\t\t\tself.log_status(\"Failed\")\n\n\tdef log_status(self, status):\n\t\t# log file\n\t\tfrappe.logger(\"scheduler\").info(f\"Scheduled Job {status}: {self.method} for {frappe.local.site}\")\n\t\tself.update_scheduler_log(status)\n\n\tdef update_scheduler_log(self, status):\n\t\tif not self.create_log:\n\t\t\t# self.get_next_execution will work properly iff self.last_execution is properly set\n\t\t\tif self.frequency == \"All\" and status == \"Start\":\n\t\t\t\tself.db_set(\"last_execution\", now_datetime(), update_modified=False)\n\t\t\t\tfrappe.db.commit()\n\t\t\treturn\n\t\tif not self.scheduler_log:\n\t\t\tself.scheduler_log = frappe.get_doc(\n\t\t\t\tdict(doctype=\"Scheduled Job Log\", scheduled_job_type=self.name)\n\t\t\t).insert(ignore_permissions=True)\n\t\tself.scheduler_log.db_set(\"status\", status)\n\t\tif status == \"Failed\":\n\t\t\tself.scheduler_log.db_set(\"details\", frappe.get_traceback(with_context=True))\n\t\tif status == \"Start\":\n\t\t\tself.db_set(\"last_execution\", now_datetime(), update_modified=False)\n\t\tfrappe.db.commit()\n\n\tdef get_queue_name(self):\n\t\treturn \"long\" if (\"Long\" in self.frequency) else \"default\"\n\n\tdef on_trash(self):\n\t\tfrappe.db.delete(\"Scheduled Job Log\", {\"scheduled_job_type\": self.name})\n\n\n@frappe.whitelist()\ndef execute_event(doc: str):\n\tfrappe.only_for(\"System Manager\")\n\tdoc = json.loads(doc)\n\tfrappe.get_doc(\"Scheduled Job Type\", doc.get(\"name\")).enqueue(force=True)\n\treturn doc\n\n\ndef run_scheduled_job(job_type: str):\n\t\"\"\"This is a wrapper function that runs a hooks.scheduler_events method\"\"\"\n\ttry:\n\t\tfrappe.get_doc(\"Scheduled Job Type\", dict(method=job_type)).execute()\n\texcept Exception:\n\t\tprint(frappe.get_traceback())\n\n\ndef sync_jobs(hooks: dict | None = None):\n\tfrappe.reload_doc(\"core\", \"doctype\", \"scheduled_job_type\")\n\tscheduler_events = hooks or frappe.get_hooks(\"scheduler_events\")\n\tall_events = insert_events(scheduler_events)\n\tclear_events(all_events)\n\n\ndef insert_events(scheduler_events: dict) -> list:\n\tcron_jobs, event_jobs = [], []\n\tfor event_type in scheduler_events:\n\t\tevents = scheduler_events.get(event_type)\n\t\tif isinstance(events, dict):\n\t\t\tcron_jobs += insert_cron_jobs(events)\n\t\telse:\n\t\t\t# hourly, daily etc\n\t\t\tevent_jobs += insert_event_jobs(events, event_type)\n\treturn cron_jobs + event_jobs\n\n\ndef insert_cron_jobs(events: dict) -> list:\n\tcron_jobs = []\n\tfor cron_format in events:\n\t\tfor event in events.get(cron_format):\n\t\t\tcron_jobs.append(event)\n\t\t\tinsert_single_event(\"Cron\", event, cron_format)\n\treturn cron_jobs\n\n\ndef insert_event_jobs(events: list, event_type: str) -> list:\n\tevent_jobs = []\n\tfor event in events:\n\t\tevent_jobs.append(event)\n\t\tfrequency = event_type.replace(\"_\", \" \").title()\n\t\tinsert_single_event(frequency, event)\n\treturn event_jobs\n\n\ndef insert_single_event(frequency: str, event: str, cron_format: str | None = None):\n\tcron_expr = {\"cron_format\": cron_format} if cron_format else {}\n\n\ttry:\n\t\tfrappe.get_attr(event)\n\texcept Exception as e:\n\t\tclick.secho(f\"{event} is not a valid method: {e}\", fg=\"yellow\")\n\n\tdoc = frappe.get_doc(\n\t\t{\n\t\t\t\"doctype\": \"Scheduled Job Type\",\n\t\t\t\"method\": event,\n\t\t\t\"cron_format\": cron_format,\n\t\t\t\"frequency\": frequency,\n\t\t}\n\t)\n\n\tif not frappe.db.exists(\"Scheduled Job Type\", {\"method\": event, \"frequency\": frequency, **cron_expr}):\n\t\tsavepoint = \"scheduled_job_type_creation\"\n\t\ttry:\n\t\t\tfrappe.db.savepoint(savepoint)\n\t\t\tdoc.insert()\n\t\texcept frappe.DuplicateEntryError:\n\t\t\tfrappe.db.rollback(save_point=savepoint)\n\t\t\tdoc.delete()\n\t\t\tdoc.insert()\n\n\ndef clear_events(all_events: list):\n\tfor event in frappe.get_all(\"Scheduled Job Type\", fields=[\"name\", \"method\", \"server_script\"]):\n\t\tis_server_script = event.server_script\n\t\tis_defined_in_hooks = event.method in all_events\n\n\t\tif not (is_defined_in_hooks or is_server_script):\n\t\t\tfrappe.delete_doc(\"Scheduled Job Type\", event.name)\n", "path": "frappe/core/doctype/scheduled_job_type/scheduled_job_type.py"}]}