:param session: SQLAlchemy ORM Session
    :meta private:
    """
    dagrun = task_instance.get_previous_dagrun(state, session=session)
    if dagrun is None:
        return None
    return dagrun.get_task_instance(task_instance.task_id, session=session)
class TaskInstance(Base, LoggingMixin):
    """
    Task instances store the state of a task instance.
    This table is the authority and single source of truth around what tasks
    have run and the state they are in.
    The SqlAlchemy model doesn't have a SqlAlchemy foreign key to the task or
    dag model deliberately to have more control over transactions.
    Database transactions on this table should insure double triggers and
    any confusion around what task instances are or aren't ready to run
    even while multiple schedulers may be firing task instances.
    A value of -1 in map_index represents any of: a TI without mapped tasks;
    a TI with mapped tasks that has yet to be expanded (state=pending);
    a TI with mapped tasks that expanded to an empty list (state=skipped).
    """
    __tablename__ = "task_instance"
    task_id = Column(StringID(), primary_key=True, nullable=False)
    dag_id = Column(StringID(), primary_key=True, nullable=False)
    run_id = Column(StringID(), primary_key=True, nullable=False)
    map_index = Column(Integer, primary_key=True, nullable=False, server_default=text("-1"))
    start_date = Column(UtcDateTime)
    end_date = Column(UtcDateTime)
    duration = Column(Float)
    state = Column(String(20))