Create new databases from the ORM

[ephraimbuddy]

This PR opens up to creating new databases from the ORM instead of going through the migration files.

airflow db init creates the new db.

Depends on ##24044

[ephraimbuddy]

Failing tests will be fixed when #24044 is merged. Also, I will need to update the migration in #24044 to add conditional index and unique on task_instance table for mssql/mysql

[norm]

This is not a prompt to change anything, merely an observation — it feels weird to create an instance of the webserver to init the DB.

[ephraimbuddy]

This is not a prompt to change anything, merely an observation — it feels weird to create an instance of the webserver to init the DB.

I have moved the creation of the session into the webserver. This means DB initialization would not create the session table but starting up the webserver would create the session table. This makes sense since the session table is from flask_session and not part of airflow.

[uranusjr]

SQLite is failing with

  File "/opt/airflow/airflow/migrations/versions/0106_2_3_0_update_migration_for_fab_tables_to_add_missing_constraints.py", line 82, in downgrade
    batch_op.drop_constraint('ab_view_menu_name_uq', type_='unique')
  File "/usr/local/lib/python3.7/contextlib.py", line 119, in __exit__
    next(self.gen)
  File "/usr/local/lib/python3.7/site-packages/alembic/operations/base.py", line 376, in batch_alter_table
    impl.flush()
  File "/usr/local/lib/python3.7/site-packages/alembic/operations/batch.py", line 142, in flush
    fn(*arg, **kw)
  File "/usr/local/lib/python3.7/site-packages/alembic/operations/batch.py", line 680, in drop_constraint
    raise ValueError("No such constraint: '%s'" % const.name)
ValueError: No such constraint: 'ab_view_menu_name_uq'

[ashb]

Given the conditionals we need to have in initdb it might be worth it if we re-factor it to be:

@provide_session
def initdb(session: Session = NEW_SESSION):
    """Initialize Airflow database."""
    if db_exists:
        _create_db_from_orm(session=session)
    else:
        upgradedb(session=session)

    if conf.getboolean('database', 'LOAD_DEFAULT_CONNECTIONS'):
        create_default_connections(session=session)

And that way upgradedb can call this new function.

[uranusjr]

Generally lgtm, just one small-ish thing