cls,
        dag_id: str,
        run_id: str,
        task_id: str,
        map_index: int,
        select_columns: bool = False,
        lock_for_update: bool = False,
        session: Session = NEW_SESSION,
    ) -> TaskInstance | TaskInstancePydantic | None:
        query = (
            session.query(*TaskInstance.__table__.columns) if select_columns else session.query(TaskInstance)
        )
        query = query.filter_by(
            dag_id=dag_id,
            run_id=run_id,
            task_id=task_id,
            map_index=map_index,
        )
        if lock_for_update:
            for attempt in run_with_db_retries(logger=cls.logger()):
                with attempt:
                    return query.with_for_update().one_or_none()
        else:
            return query.one_or_none()
        return None
    @provide_session
    def refresh_from_db(self, session: Session = NEW_SESSION, lock_for_update: bool = False) -> None: