uranusjr commented on code in PR #29355:
URL: https://github.com/apache/airflow/pull/29355#discussion_r1100345687
##########
airflow/models/taskinstance.py:
##########
@@ -495,42 +503,59 @@ def __init__(
)
# make sure we have a localized execution_date stored in UTC
if execution_date and not timezone.is_localized(execution_date):
- self.log.warning(
+ ti.log.warning(
"execution date %s has no timezone information. Using
default from dag or system",
execution_date,
)
- if self.task.has_dag():
+ if ti.task.has_dag():
if TYPE_CHECKING:
- assert self.task.dag
- execution_date = timezone.make_aware(execution_date,
self.task.dag.timezone)
+ assert ti.task.dag
+ execution_date = timezone.make_aware(execution_date,
ti.task.dag.timezone)
else:
execution_date = timezone.make_aware(execution_date)
execution_date = timezone.convert_to_utc(execution_date)
with create_session() as session:
run_id = (
session.query(DagRun.run_id)
- .filter_by(dag_id=self.dag_id,
execution_date=execution_date)
+ .filter_by(dag_id=ti.dag_id, execution_date=execution_date)
.scalar()
)
if run_id is None:
raise DagRunNotFound(
- f"DagRun for {self.dag_id!r} with date
{execution_date} not found"
+ f"DagRun for {ti.dag_id!r} with date {execution_date}
not found"
) from None
- self.run_id = run_id
+ ti.run_id = run_id
- self.try_number = 0
- self.max_tries = self.task.retries
- self.unixname = getuser()
+ ti.try_number = 0
+ ti.max_tries = ti.task.retries
+ ti.unixname = getuser()
if state:
- self.state = state
- self.hostname = ""
+ ti.state = state
+ ti.hostname = ""
# Is this TaskInstance being currently running within `airflow tasks
run --raw`.
# Not persisted to the database so only valid for the current process
- self.raw = False
+ ti.raw = False
# can be changed when calling 'run'
- self.test_mode = False
+ ti.test_mode = False
+ return ti
+
+ @staticmethod
+ def from_dict(ti_dict: dict[str, Any]) -> TaskInstance:
+ """Create TaskInstance from dictionary."""
+ ti = TaskInstance()
+ ti.__dict__ = ti_dict.copy()
+ ti.init_on_load()
+ return ti
+
+ def to_dict(self) -> dict[str, Any]:
+ ti_dict = self.__dict__.copy()
+ ti_dict.pop("_log", None)
+ ti_dict.pop("test_mode", None)
+ ti_dict.pop("dag_run", None)
Review Comment:
Basically you’d list out what fields you expect to be in the dict in the
pre-commit check, and compare it with `__dict__` minus these explicitly
excluded fields. So if `__dict__` is changed, someone must consciously decide
whether the new field should go into the dict (by adding it to the list
maintained in the check), or be excluded (by adding it to the exclude fields).
This can also be done in a test. (There are several serialization tests that
do exactly this.)
--
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]
