Source code for airflow.models.taskinstancekey

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
from __future__ import annotations

from typing import NamedTuple

[docs]class TaskInstanceKey(NamedTuple): """Key used to identify task instance."""
[docs] dag_id: str
[docs] task_id: str
[docs] run_id: str
[docs] try_number: int = 1
[docs] map_index: int = -1
[docs] def primary(self) -> tuple[str, str, str, int]: """Return task instance primary key part of the key.""" return self.dag_id, self.task_id, self.run_id, self.map_index
[docs] def reduced(self) -> TaskInstanceKey: """Remake the key by subtracting 1 from try number to match in memory information.""" return TaskInstanceKey( self.dag_id, self.task_id, self.run_id, max(1, self.try_number - 1), self.map_index )
[docs] def with_try_number(self, try_number: int) -> TaskInstanceKey: """Return TaskInstanceKey with provided ``try_number``.""" return TaskInstanceKey(self.dag_id, self.task_id, self.run_id, try_number, self.map_index)
[docs] def key(self) -> TaskInstanceKey: """For API-compatibly with TaskInstance. Returns self """ return self

Was this entry helpful?