### Discussed in https://github.com/apache/airflow/discussions/18408

<div type='discussions-op-text'>

<sup>Originally posted by **ziplokk1** September 21, 2021</sup>
Hey folks,

I've come across an issue where when attempting to run migrations against a clean database, but with existing dags defined in my dags folder, there's a silent failure and the migration exits with the `airflow` command usage prompt saying "invalid command 'db'". 

I got to digging and the culprit was the [following line](https://github.com/apache/airflow/blob/main/airflow/migrations/versions/cc1e65623dc7_add_max_tries_column_to_task_instance.py#L73).

I'm almost positive that this has to do with an erroneous `.py` file in my dags folder (I have yet to dig deeper into the issue), but this begged the question of why is the migration is reading the dags to migrate existing data in the database in the first place? Shouldn't the migration simply migrate the data regardless of the state of the dag files themselves?

I was able to fix the issue by changing the line in the link above to `dagbag = DagBag(read_dags_from_db=True)` but I'd like to see if there was any context behind reading the dag file instead of the database before opening a PR.</div>
airflow/migrations/versions/cc1e65623dc7_add_max_tries_column_to_task_instance.py
engine = settings.engine
    if engine.dialect.has_table(engine, 'task_instance'):
        connection = op.get_bind()
        sessionmaker = sa.orm.sessionmaker()
        session = sessionmaker(bind=connection)
        dagbag = DagBag(settings.DAGS_FOLDER)
        query = session.query(sa.func.count(TaskInstance.max_tries)).filter(TaskInstance.max_tries != -1)
        while query.scalar():
            tis = session.query(TaskInstance).filter(TaskInstance.max_tries != -1).limit(BATCH_SIZE).all()
            for ti in tis:
                dag = dagbag.get_dag(ti.dag_id)
                if not dag or not dag.has_task(ti.task_id):
                    ti.try_number = 0
                else:
                    task = dag.get_task(ti.task_id)
                    
                    
                    
                    
                    ti.try_number = max(0, task.retries - (ti.max_tries - ti.try_number))
                ti.max_tries = -1
                session.merge(ti)
            session.commit()
        session.commit()
    op.drop_column('task_instance', 'max_tries')