Detect and fail any command when db is not migrated

[potiuk]

Description

We had recently a number of questions from users who did not run airflow db upgrade and run airflow and it was failing because of missing columns etc.

It was not entirely clear for new users that they should do it and I am fixing it in #18282 , however I think we should do better

Use case/motivation

I think we should check what migration is currently applied and fail any Airflow command that requires the DB hard with clear information that the DB is not migrated and what to do.

This should be rather easy, I think. Alembic keeps track of the migration and I believe it has already all the tools to determine "current migration" vs. "expected migration".

Related issues

#18282

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[potiuk]

cc:@kaxil - WDYT? I do not know Alembic that well, but from what I understand, I believe this would be entirely possible to perform such a check and clearly raise error "Please migrate the db with airflow db upgrade " in case the migration revision is different than expected?

[bhavaniravi]

Yes, it's possible in the alembic. Check this SO answer

A similar airflow snippet would look like, used check_migrations as a template

def compare_migrations():
    from alembic.runtime.migration import MigrationContext
    from alembic.script import ScriptDirectory

    config = _get_alembic_config()
    script_ = ScriptDirectory.from_config(config)
    with settings.engine.connect() as connection:
        context = MigrationContext.configure(connection)
        if context.get_current_revision() != script_.get_current_head():
            raise exceptions.DatabaseIsNotUpToDate('Upgrade the database.')

Do we do this check every time we start the webserver or scheduler?

[kaxil]

Yea we have something similar that we already use in the Helm Chart:

airflow/airflow/utils/db.py

Lines 622 to 649 in 11621ce

	def check_migrations(timeout):
	"""
	Function to wait for all airflow migrations to complete.
	:param timeout: Timeout for the migration in seconds
	:return: None
	"""
	from alembic.runtime.migration import MigrationContext
	from alembic.script import ScriptDirectory
	config = _get_alembic_config()
	script_ = ScriptDirectory.from_config(config)
	with settings.engine.connect() as connection:
	context = MigrationContext.configure(connection)
	ticker = 0
	while True:
	source_heads = set(script_.get_heads())
	db_heads = set(context.get_current_heads())
	if source_heads == db_heads:
	break
	if ticker >= timeout:
	raise TimeoutError(
	f"There are still unapplied migrations after {ticker} seconds. "
	f"Migration Head(s) in DB: {db_heads} | Migration Head(s) in Source Code: {source_heads}"
	)
	ticker += 1
	time.sleep(1)
	log.info('Waiting for migrations... %s second(s)', ticker)

I will add that to the components

[uranusjr]

One minor thought (aside from we obviously can’t show this error for airflow db upgrade) is maybe we need to provide an override mechanism for local development, it’s occasionally useful knowingly run things without full migration (to fix an existing migration for example).

[potiuk]

One minor thought (aside from we obviously can’t show this error for airflow db upgrade) is maybe we need to provide an override mechanism for local development, it’s occasionally useful knowingly run things without full migration (to fix an existing migration for example).

Yep. Good call. And yeah. I was toying with the thought of enabling the check for airflow db upgrade just for the fun of it.
I was thinking about adding it first and seeing if someone will find it during the review ;). But you spoiled my sneaky plan TP

[uranusjr]

Another one #18368