diff --git a/frappe/app.py b/frappe/app.py
index 03414646ef..6b16464d2c 100644
--- a/frappe/app.py
+++ b/frappe/app.py
@@ -41,9 +41,6 @@ def application(request: Request):
 
 		init_request(request)
 
-		frappe.recorder.record()
-		frappe.monitor.start()
-		frappe.rate_limiter.apply()
 		frappe.api.validate_auth()
 
 		if request.method == "OPTIONS":
@@ -74,15 +71,14 @@ def application(request: Request):
 		response = handle_exception(e)
 
 	else:
-		rollback = after_request(rollback)
+		rollback = sync_database(rollback)
 
 	finally:
 		if request.method in UNSAFE_HTTP_METHODS and frappe.db and rollback:
 			frappe.db.rollback()
 
-		frappe.rate_limiter.update()
-		frappe.monitor.stop(response)
-		frappe.recorder.dump()
+		for after_request_task in frappe.get_hooks("after_request"):
+			frappe.call(after_request_task, response=response, request=request)
 
 		log_request(request, response)
 		process_response(response)
@@ -119,6 +115,9 @@ def init_request(request):
 	if request.method != "OPTIONS":
 		frappe.local.http_request = HTTPRequest()
 
+	for before_request_task in frappe.get_hooks("before_request"):
+		frappe.call(before_request_task)
+
 
 def setup_read_only_mode():
 	"""During maintenance_mode reads to DB can still be performed to reduce downtime. This
@@ -318,7 +317,7 @@ def handle_exception(e):
 	return response
 
 
-def after_request(rollback):
+def sync_database(rollback: bool) -> bool:
 	# if HTTP method would change server state, commit if necessary
 	if (
 		frappe.db
@@ -332,9 +331,8 @@ def after_request(rollback):
 		rollback = False
 
 	# update session
-	if getattr(frappe.local, "session_obj", None):
-		updated_in_db = frappe.local.session_obj.update()
-		if updated_in_db:
+	if session := getattr(frappe.local, "session_obj", None):
+		if session.update():
 			frappe.db.commit()
 			rollback = False
 
diff --git a/frappe/hooks.py b/frappe/hooks.py
index aa6d6f00c7..317439c358 100644
--- a/frappe/hooks.py
+++ b/frappe/hooks.py
@@ -393,3 +393,20 @@ ignore_links_on_delete = [
 	"Integration Request",
 	"Unhandled Email",
 ]
+
+# Request Hooks
+before_request = [
+	"frappe.recorder.record",
+	"frappe.monitor.start",
+	"frappe.rate_limiter.apply",
+]
+after_request = ["frappe.rate_limiter.update", "frappe.monitor.stop", "frappe.recorder.dump"]
+
+# Background Job Hooks
+before_job = [
+	"frappe.monitor.start",
+]
+after_job = [
+	"frappe.monitor.stop",
+	"frappe.utils.file_lock.release_document_locks",
+]
diff --git a/frappe/public/js/frappe/form/form.js b/frappe/public/js/frappe/form/form.js
index 001279f394..d1fe443f7a 100644
--- a/frappe/public/js/frappe/form/form.js
+++ b/frappe/public/js/frappe/form/form.js
@@ -406,7 +406,10 @@ frappe.ui.form.Form = class FrappeForm {
 
 			// read only (workflow)
 			this.read_only = frappe.workflow.is_read_only(this.doctype, this.docname);
-			if (this.read_only) this.set_read_only(true);
+			if (this.read_only) {
+				this.set_read_only(true);
+				frappe.show_alert(__("This form is not editable due to a Workflow."));
+			}
 
 			// check if doctype is already open
 			if (!this.opendocs[this.docname]) {
diff --git a/frappe/tests/test_api.py b/frappe/tests/test_api.py
index 1085f4c39e..c345d8fbcf 100644
--- a/frappe/tests/test_api.py
+++ b/frappe/tests/test_api.py
@@ -2,6 +2,7 @@ import sys
 from contextlib import contextmanager
 from random import choice
 from threading import Thread
+from time import time
 from unittest.mock import patch
 
 import requests
@@ -306,3 +307,36 @@ class TestReadOnlyMode(FrappeAPITestCase):
 		response = self.post(self.REQ_PATH, {"description": frappe.mock("paragraph"), "sid": self.sid})
 		self.assertEqual(response.status_code, 503)
 		self.assertEqual(response.json["exc_type"], "InReadOnlyMode")
+
+
+class TestWSGIApp(FrappeAPITestCase):
+	def test_request_hooks(self):
+		self.addCleanup(lambda: _test_REQ_HOOK.clear())
+		get_hooks = frappe.get_hooks
+
+		def patch_request_hooks(event: str, *args, **kwargs):
+			patched_hooks = {
+				"before_request": ["frappe.tests.test_api.before_request"],
+				"after_request": ["frappe.tests.test_api.after_request"],
+			}
+			if event not in patched_hooks:
+				return get_hooks(event, *args, **kwargs)
+			return patched_hooks[event]
+
+		with patch("frappe.get_hooks", patch_request_hooks):
+			self.assertIsNone(_test_REQ_HOOK.get("before_request"))
+			self.assertIsNone(_test_REQ_HOOK.get("after_request"))
+			res = self.get("/api/method/ping")
+			self.assertEqual(res.json, {"message": "pong"})
+			self.assertLess(_test_REQ_HOOK.get("before_request"), _test_REQ_HOOK.get("after_request"))
+
+
+_test_REQ_HOOK = {}
+
+
+def before_request(*args, **kwargs):
+	_test_REQ_HOOK["before_request"] = time()
+
+
+def after_request(*args, **kwargs):
+	_test_REQ_HOOK["after_request"] = time()
diff --git a/frappe/tests/test_background_jobs.py b/frappe/tests/test_background_jobs.py
index 1b6680eb9b..72660128cf 100644
--- a/frappe/tests/test_background_jobs.py
+++ b/frappe/tests/test_background_jobs.py
@@ -1,11 +1,19 @@
 import time
+from contextlib import contextmanager
+from unittest.mock import patch
 
 from rq import Queue
 
 import frappe
 from frappe.core.doctype.rq_job.rq_job import remove_failed_jobs
 from frappe.tests.utils import FrappeTestCase
-from frappe.utils.background_jobs import generate_qname, get_redis_conn
+from frappe.utils.background_jobs import (
+	RQ_JOB_FAILURE_TTL,
+	RQ_RESULTS_TTL,
+	execute_job,
+	generate_qname,
+	get_redis_conn,
+)
 
 
 class TestBackgroundJobs(FrappeTestCase):
@@ -44,6 +52,79 @@ class TestBackgroundJobs(FrappeTestCase):
 		# lesser is earlier
 		self.assertTrue(high_priority_job.get_position() < low_priority_job.get_position())
 
+	def test_enqueue_call(self):
+		with patch.object(Queue, "enqueue_call") as mock_enqueue_call:
+			frappe.enqueue(
+				"frappe.handler.ping",
+				queue="short",
+				timeout=300,
+				kwargs={"site": frappe.local.site},
+			)
+
+			mock_enqueue_call.assert_called_once_with(
+				execute_job,
+				on_success=None,
+				on_failure=None,
+				timeout=300,
+				kwargs={
+					"site": frappe.local.site,
+					"user": "Administrator",
+					"method": "frappe.handler.ping",
+					"event": None,
+					"job_name": "frappe.handler.ping",
+					"is_async": True,
+					"kwargs": {"kwargs": {"site": frappe.local.site}},
+				},
+				at_front=False,
+				failure_ttl=RQ_JOB_FAILURE_TTL,
+				result_ttl=RQ_RESULTS_TTL,
+			)
+
+	def test_job_hooks(self):
+		self.addCleanup(lambda: _test_JOB_HOOK.clear())
+		with freeze_local() as locals, frappe.init_site(locals.site), patch(
+			"frappe.get_hooks", patch_job_hooks
+		):
+			frappe.connect()
+			self.assertIsNone(_test_JOB_HOOK.get("before_job"))
+			r = execute_job(
+				site=frappe.local.site,
+				user="Administrator",
+				method="frappe.handler.ping",
+				event=None,
+				job_name="frappe.handler.ping",
+				is_async=True,
+				kwargs={},
+			)
+			self.assertEqual(r, "pong")
+			self.assertLess(_test_JOB_HOOK.get("before_job"), _test_JOB_HOOK.get("after_job"))
+
 
 def fail_function():
 	return 1 / 0
+
+
+_test_JOB_HOOK = {}
+
+
+def before_job(*args, **kwargs):
+	_test_JOB_HOOK["before_job"] = time.time()
+
+
+def after_job(*args, **kwargs):
+	_test_JOB_HOOK["after_job"] = time.time()
+
+
+@contextmanager
+def freeze_local():
+	locals = frappe.local
+	frappe.local = frappe.Local()
+	yield locals
+	frappe.local = locals
+
+
+def patch_job_hooks(event: str):
+	return {
+		"before_job": ["frappe.tests.test_background_jobs.before_job"],
+		"after_job": ["frappe.tests.test_background_jobs.after_job"],
+	}[event]
diff --git a/frappe/utils/background_jobs.py b/frappe/utils/background_jobs.py
index 040a57cc11..fed822700c 100755
--- a/frappe/utils/background_jobs.py
+++ b/frappe/utils/background_jobs.py
@@ -153,6 +153,7 @@ def run_doc_method(doctype, name, doc_method, **kwargs):
 
 def execute_job(site, method, event, job_name, kwargs, user=None, is_async=True, retry=0):
 	"""Executes job in a worker, performs commit/rollback and logs if there is any error"""
+	retval = None
 	if is_async:
 		frappe.connect(site)
 		if os.environ.get("CI"):
@@ -167,9 +168,11 @@ def execute_job(site, method, event, job_name, kwargs, user=None, is_async=True,
 	else:
 		method_name = cstr(method.__name__)
 
-	frappe.monitor.start("job", method_name, kwargs)
+	for before_job_task in frappe.get_hooks("before_job"):
+		frappe.call(before_job_task, method=method_name, kwargs=kwargs, transaction_type="job")
+
 	try:
-		method(**kwargs)
+		retval = method(**kwargs)
 
 	except (frappe.db.InternalError, frappe.RetryBackgroundJobError) as e:
 		frappe.db.rollback()
@@ -200,14 +203,12 @@ def execute_job(site, method, event, job_name, kwargs, user=None, is_async=True,
 
 	else:
 		frappe.db.commit()
+		return retval
 
 	finally:
-		# background job hygiene: release file locks if unreleased
-		# if this breaks something, move it to failed jobs alone - gavin@frappe.io
-		for doc in frappe.local.locked_documents:
-			doc.unlock()
+		for after_job_task in frappe.get_hooks("after_job"):
+			frappe.call(after_job_task, method=method_name, kwargs=kwargs, result=retval)
 
-		frappe.monitor.stop()
 		if is_async:
 			frappe.destroy()
 
diff --git a/frappe/utils/boilerplate.py b/frappe/utils/boilerplate.py
index ee75c672a8..1cd57f4695 100644
--- a/frappe/utils/boilerplate.py
+++ b/frappe/utils/boilerplate.py
@@ -458,6 +458,15 @@ app_license = "{app_license}"
 
 # ignore_links_on_delete = ["Communication", "ToDo"]
 
+# Request Events
+# ----------------
+# before_request = ["{app_name}.utils.before_request"]
+# after_request = ["{app_name}.utils.after_request"]
+
+# Job Events
+# ----------
+# before_job = ["{app_name}.utils.before_job"]
+# after_job = ["{app_name}.utils.after_job"]
 
 # User Data Protection
 # --------------------
diff --git a/frappe/utils/file_lock.py b/frappe/utils/file_lock.py
index 5be89c42f6..60d8baec5d 100644
--- a/frappe/utils/file_lock.py
+++ b/frappe/utils/file_lock.py
@@ -11,7 +11,7 @@ Use `frappe.utils.synchroniztion.filelock` for process synchroniztion.
 import os
 from time import time
 
-from frappe import _
+import frappe
 from frappe.utils import get_site_path, touch_file
 
 LOCKS_DIR = "locks"
@@ -62,3 +62,9 @@ def get_lock_path(name):
 	name = name.lower()
 	lock_path = get_site_path(LOCKS_DIR, name + ".lock")
 	return lock_path
+
+
+def release_document_locks():
+	"""Unlocks all documents that were locked by the current context."""
+	for doc in frappe.local.locked_documents:
+		doc.unlock()
diff --git a/frappe/website/doctype/blog_post/blog_post.js b/frappe/website/doctype/blog_post/blog_post.js
index 0266587f2e..5f7268d074 100644
--- a/frappe/website/doctype/blog_post/blog_post.js
+++ b/frappe/website/doctype/blog_post/blog_post.js
@@ -7,6 +7,8 @@ frappe.ui.form.on("Blog Post", {
 			frm.set_df_property("hide_cta", "hidden", !value);
 		});
 
+		frm.trigger("add_publish_button");
+
 		generate_google_search_preview(frm);
 	},
 	title: function (frm) {
@@ -30,6 +32,12 @@ frappe.ui.form.on("Blog Post", {
 			});
 		}
 	},
+	add_publish_button(frm) {
+		frm.add_custom_button(frm.doc.published ? __("Unpublish") : __("Publish"), () => {
+			frm.set_value("published", !frm.doc.published);
+			frm.save();
+		});
+	},
 });
 
 function generate_google_search_preview(frm) {
diff --git a/frappe/website/doctype/blog_post/blog_post.json b/frappe/website/doctype/blog_post/blog_post.json
index 8b5ab54ba8..d41b344464 100644
--- a/frappe/website/doctype/blog_post/blog_post.json
+++ b/frappe/website/doctype/blog_post/blog_post.json
@@ -53,6 +53,7 @@
    "default": "0",
    "fieldname": "published",
    "fieldtype": "Check",
+   "hidden": 1,
    "label": "Published"
   },
   {
@@ -215,7 +216,7 @@
  "is_published_field": "published",
  "links": [],
  "make_attachments_public": 1,
- "modified": "2022-10-18 10:09:10.550734",
+ "modified": "2023-02-17 11:31:32.223524",
  "modified_by": "Administrator",
  "module": "Website",
  "name": "Blog Post",