https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
self.state = state
        if queued_at is NOTSET:
            self.queued_at = timezone.utcnow() if state == DagRunState.QUEUED else None
        else:
            self.queued_at = queued_at
        self.run_type = run_type
        self.dag_hash = dag_hash
        self.creating_job_id = creating_job_id
        self.clear_number = 0
        super().__init__()
    def __repr__(self):
        return (
            f"<DagRun {self.dag_id} @ {self.execution_date}: {self.run_id}, state:{self.state}, "
            f"queued_at: {self.queued_at}. externally triggered: {self.external_trigger}>"
        )
    @validates("run_id")
    def validate_run_id(self, key: str, run_id: str) -> str | None:
        if not run_id:
            return None
        regex = airflow_conf.get("scheduler", "allowed_run_id_pattern")
        if not re2.match(regex, run_id) and not re2.match(RUN_ID_REGEX, run_id):
            raise ValueError(
                f"The run_id provided '{run_id}' does not match the pattern '{regex}' or '{RUN_ID_REGEX}'"
            )
        return run_id
    @property
    def stats_tags(self) -> dict[str, str]:
        return prune_dict({"dag_id": self.dag_id, "run_type": self.run_type})
    @property
    def logical_date(self) -> datetime: