Add pre/post execution hooks
diff --git a/airflow/models/baseoperator.py b/airflow/models/baseoperator.py
index a1316405b4e91..3feeb5b1e2851 100644
--- a/airflow/models/baseoperator.py
+++ b/airflow/models/baseoperator.py
@@ -81,6 +81,8 @@
 ScheduleInterval = Union[str, timedelta, relativedelta]
 
 TaskStateChangeCallback = Callable[[Context], None]
+TaskPreExecuteHook = Callable[[Context], None]
+TaskPostExecuteHook = Callable[[Context, Any], None]
 
 T = TypeVar('T', bound=Callable)
 
@@ -347,6 +349,14 @@ class derived from this one results in the creation of a task object,
     :param on_success_callback: much like the ``on_failure_callback`` except
         that it is executed when the task succeeds.
     :type on_success_callback: TaskStateChangeCallback
+    :param pre_execute: a function to be called immediately before task
+        execution, receiving a context dictionary; raising an exception will
+        prevent the task from being executed.
+    :type pre_execute: TaskPreExecuteHook
+    :param post_execute: a function to be called immediately after task
+        execution, receiving a context dictionary and task result; raising an
+        exception will prevent the task from succeeding.
+    :type post_execute: TaskPostExecuteHook
     :param trigger_rule: defines the rule by which dependencies are applied
         for the task to get triggered. Options are:
         ``{ all_success | all_failed | all_done | one_success |
@@ -488,6 +498,8 @@ def __init__(
         on_failure_callback: Optional[TaskStateChangeCallback] = None,
         on_success_callback: Optional[TaskStateChangeCallback] = None,
         on_retry_callback: Optional[TaskStateChangeCallback] = None,
+        pre_execute: Optional[TaskPreExecuteHook] = None,
+        post_execute: Optional[TaskPostExecuteHook] = None,
         trigger_rule: str = TriggerRule.ALL_SUCCESS,
         resources: Optional[Dict] = None,
         run_as_user: Optional[str] = None,
@@ -599,6 +611,8 @@ def __init__(
         self.on_failure_callback = on_failure_callback
         self.on_success_callback = on_success_callback
         self.on_retry_callback = on_retry_callback
+        self._pre_execute_hook = pre_execute
+        self._post_execute_hook = post_execute
 
         if isinstance(retry_delay, timedelta):
             self.retry_delay = retry_delay
@@ -960,6 +974,8 @@ def global_operator_extra_link_dict(self) -> Dict[str, Any]:
     @prepare_lineage
     def pre_execute(self, context: Any):
         """This hook is triggered right before self.execute() is called."""
+        if self._pre_execute_hook is not None:
+            self._pre_execute_hook(context)
 
     def execute(self, context: Any):
         """
@@ -977,6 +993,8 @@ def post_execute(self, context: Any, result: Any = None):
         It is passed the execution context and any results returned by the
         operator.
         """
+        if self._post_execute_hook is not None:
+            self._post_execute_hook(context, result)
 
     def on_kill(self) -> None:
         """
diff --git a/airflow/operators/subdag.py b/airflow/operators/subdag.py
index 62b9b776d6b32..da83bee03ad1f 100644
--- a/airflow/operators/subdag.py
+++ b/airflow/operators/subdag.py
@@ -156,6 +156,7 @@ def _reset_dag_run_and_task_instances(self, dag_run, execution_date):
             session.commit()
 
     def pre_execute(self, context):
+        super().pre_execute(context)
         execution_date = context['execution_date']
         dag_run = self._get_dagrun(execution_date)
 
@@ -184,6 +185,7 @@ def poke(self, context):
         return dag_run.state != State.RUNNING
 
     def post_execute(self, context, result=None):
+        super().post_execute(context)
         execution_date = context['execution_date']
         dag_run = self._get_dagrun(execution_date=execution_date)
         self.log.info("Execution finished. State is %s", dag_run.state)
diff --git a/tests/models/test_baseoperator.py b/tests/models/test_baseoperator.py
index ce848db0ed4c7..bf3ea4a7f4481 100644
--- a/tests/models/test_baseoperator.py
+++ b/tests/models/test_baseoperator.py
@@ -541,6 +541,30 @@ def test_warnings_are_properly_propagated(self):
             # where the deprecated class was used
             assert warning.filename == __file__
 
+    def test_pre_execute_hook(self):
+        called = False
+
+        def hook(context):
+            nonlocal called
+            called = True
+
+        op = DummyOperator(task_id="test_task", pre_execute=hook)
+        op_copy = op.prepare_for_execution()
+        op_copy.pre_execute({})
+        assert called
+
+    def test_post_execute_hook(self):
+        called = False
+
+        def hook(context, result):
+            nonlocal called
+            called = True
+
+        op = DummyOperator(task_id="test_task", post_execute=hook)
+        op_copy = op.prepare_for_execution()
+        op_copy.post_execute({})
+        assert called
+
 
 class CustomOp(DummyOperator):
     template_fields = ("field", "field2")
diff --git a/tests/serialization/test_dag_serialization.py b/tests/serialization/test_dag_serialization.py
index 826b10f6b3207..c3872165e09b1 100644
--- a/tests/serialization/test_dag_serialization.py
+++ b/tests/serialization/test_dag_serialization.py
@@ -872,6 +872,8 @@ def test_no_new_fields_added_to_base_operator(self):
             '_log': base_operator.log,
             '_outlets': [],
             '_upstream_task_ids': set(),
+            '_pre_execute_hook': None,
+            '_post_execute_hook': None,
             'depends_on_past': False,
             'do_xcom_push': True,
             'doc': None,

This adds optional operator parameters `pre_execute` and `post_execute`, both of which run synchronously immediately before and after task execution.

For example, this can be used to skip execution for a particular day of the week:
```python
from croniter import match

def execute_unless(cron_expr):
    def hook(context):
        if match(cron_expr, context["execution_date"]):
            raise AirflowSkipException()
    return hook

DummyOperator(..., pre_execute=execute_unless("* * * * sat"))
```

Related: #17545