amoghrajesh commented on code in PR #48244:
URL: https://github.com/apache/airflow/pull/48244#discussion_r2018078217
##########
airflow-core/tests/unit/serialization/test_dag_serialization.py:
##########
@@ -750,6 +748,13 @@ def validate_deserialized_task(
else:
assert serialized_task.resources == task.resources
+ # `deps` are set in the Scheduler's BaseOperator as that is where we
need to evaluate deps
+ # so only serialized tasks that are sensors should have the
ReadyToRescheduleDep.
+ if task._is_sensor:
Review Comment:
Cool yes makes sense
##########
task-sdk/src/airflow/sdk/execution_time/task_runner.py:
##########
@@ -385,16 +385,14 @@ def get_first_reschedule_date(self, context: Context) ->
datetime | None:
if TYPE_CHECKING:
assert isinstance(response, TaskRescheduleStartDate)
- start_date = response.start_date
- log.debug("First reschedule date from supervisor: %s", start_date)
-
- return start_date
+ return response.start_date
Review Comment:
Can we keep the log?
##########
task-sdk/src/airflow/sdk/definitions/baseoperator.py:
##########
@@ -1549,6 +1549,26 @@ def render_template_fields(
jinja_env = self.get_template_env()
self._do_render_template_fields(self, self.template_fields, context,
jinja_env, set())
+ def pre_execute(self, context: Any):
+ """Execute right before self.execute() is called."""
Review Comment:
Add a pass to be explicit?
##########
providers/databricks/tests/unit/databricks/sensors/test_databricks_partition.py:
##########
@@ -91,11 +91,6 @@ def test_poke(self, mock_poke, sensor_poke_result,
expected_poke_result):
mock_poke.return_value = sensor_poke_result
assert self.partition_sensor.poke({}) == expected_poke_result
- @pytest.mark.db_test
Review Comment:
Yeah!
##########
providers/http/tests/unit/http/sensors/test_http.py:
##########
@@ -317,8 +319,25 @@ def test_get_response_check(self):
)
op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE,
ignore_ti_state=True)
+ @pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="Test only for Airflow
3.0+")
+ @mock.patch("airflow.providers.http.hooks.http.Session", FakeSession)
+ def test_sensor(self, run_task):
+ sensor = HttpSensor(
+ task_id="http_sensor_check",
+ http_conn_id="http_default",
+ endpoint="/search",
+ request_params={"client": "ubuntu", "q": "airflow", "date":
"{{ds}}"},
+ headers={},
+ response_check=lambda response:
f"apache/airflow/{DEFAULT_DATE:%Y-%m-%d}" in response.text,
+ poke_interval=5,
+ timeout=15,
+ dag=self.dag,
+ )
+ run_task(sensor)
Review Comment:
Should we assert something? I know we are testing if it works, so not sure
##########
task-sdk/src/airflow/sdk/definitions/baseoperator.py:
##########
@@ -1549,6 +1549,26 @@ def render_template_fields(
jinja_env = self.get_template_env()
self._do_render_template_fields(self, self.template_fields, context,
jinja_env, set())
+ def pre_execute(self, context: Any):
+ """Execute right before self.execute() is called."""
+
+ def execute(self, context: Context) -> Any:
+ """
+ Derive when creating an operator.
+
+ The main method to execute the task. Context is the same dictionary
used as when rendering jinja templates.
+
+ Refer to get_template_context for more context.
+ """
+ raise NotImplementedError()
+
+ def post_execute(self, context: Any, result: Any = None):
+ """
+ Execute right after self.execute() is called.
+
+ It is passed the execution context and any results returned by the
operator.
+ """
Review Comment:
Add a pass to be explicit?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
