AirFlow Unit Testing not working as described in the documentation:

[camilla-storytel]

What do you see as an issue?

Hi!
I am trying to unit test my DAGs by following this link in the AirFlow docs:

import pytest

from airflow.models import DagBag


@pytest.fixture()
def dagbag():
    return DagBag()


def test_dag_loaded(dagbag):
    dag = dagbag.get_dag(dag_id="hello_world")
    assert dagbag.import_errors == {}
    assert dag is not None
    assert len(dag.tasks) == 1

However, when I try to run this with pytest, I get this error:

../env/lib/python3.10/site-packages/sqlalchemy/engine/base.py:1910:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <sqlalchemy.dialects.sqlite.pysqlite.SQLiteDialect_pysqlite object at 0x105bb66b0>, cursor = <sqlite3.Cursor object at 0x15c56bf40>
statement = 'SELECT dag.dag_id AS dag_dag_id, dag.root_dag_id AS dag_root_dag_id, dag.is_paused AS dag_is_paused, dag.is_subdag AS...end, dag.next_dagrun_create_after AS dag_next_dagrun_create_after \nFROM dag \nWHERE dag.dag_id = ?\n LIMIT ? OFFSET ?'
parameters = ('long-tail-recommender-0.0.1', 1, 0), context = <sqlalchemy.dialects.sqlite.base.SQLiteExecutionContext object at 0x15c713a30>

    def do_execute(self, cursor, statement, parameters, context=None):
>       cursor.execute(statement, parameters)
E       sqlite3.OperationalError: no such table: dag

../env/lib/python3.10/site-packages/sqlalchemy/engine/default.py:736: OperationalError

The above exception was the direct cause of the following exception:

I suspect I am lacking some local setup, but I don't really know which one and there is no further guidance in the documentation.

Solving the problem

If someone could please help me figure out why I am missing the dag table and what the proper setup would look like, I will happily open a PR to improve the documentaiton.

Anything else

This occurs every time I run pytest on the unit test example.

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

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

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[Taragolis]

You need to initialise/migrate DB first by run airflow db migrate

[camilla-storytel]

Thank you for the reply!
It looks like there is no migrate option under the db commands:

airflow db command error: argument COMMAND: invalid choice: 'migrate' (choose from 'check', 'check-migrations', 'clean', 'downgrade', 'drop-archived', 'export-archived', 'init', 'reset', 'shell', 'upgrade'), see help above.
(env) camilla.montonen@STL-D14J5637T2 pipelines % airflow db --help
Usage: airflow db [-h] COMMAND ...

Database operations

Positional Arguments:
  COMMAND
    check           Check if the database can be reached
    check-migrations
                    Check if migration have finished
    clean           Purge old records in metastore tables
    downgrade       Downgrade the schema of the metadata database.
    drop-archived   Drop archived tables created through the db clean command
    export-archived
                    Export archived data from the archive tables
    init            Initialize the metadata database
    reset           Burn down and rebuild the metadata database
    shell           Runs a shell to access the database
    upgrade         Upgrade the metadata database to latest version

Options:
  -h, --help        show this help message and exit

I tried to initialise the db using airflow db init and got this error message about missing a missing flask_session.sessions library.

(env) camilla.montonen@STL-D14J5637T2 pipelines % airflow db init
DB: sqlite:////Users/camilla.montonen/airflow/airflow.db
[2024-05-05T12:49:14.983+0200] {migration.py:216} INFO - Context impl SQLiteImpl.
[2024-05-05T12:49:14.985+0200] {migration.py:219} INFO - Will assume non-transactional DDL.
Traceback (most recent call last):
  File "/Users/camilla.montonen/backups/development/recommender-pipelines/env/bin/airflow", line 8, in <module>
    sys.exit(main())
  File "/Users/camilla.montonen/backups/development/recommender-pipelines/env/lib/python3.10/site-packages/airflow/__main__.py", line 48, in main
    args.func(args)
  File "/Users/camilla.montonen/backups/development/recommender-pipelines/env/lib/python3.10/site-packages/airflow/cli/cli_config.py", line 52, in command
    return func(*args, **kwargs)
  File "/Users/camilla.montonen/backups/development/recommender-pipelines/env/lib/python3.10/site-packages/airflow/cli/commands/db_command.py", line 37, in initdb
    db.initdb()
  File "/Users/camilla.montonen/backups/development/recommender-pipelines/env/lib/python3.10/site-packages/airflow/utils/session.py", line 76, in wrapper
    return func(*args, session=session, **kwargs)
  File "/Users/camilla.montonen/backups/development/recommender-pipelines/env/lib/python3.10/site-packages/airflow/utils/db.py", line 718, in initdb
    _create_db_from_orm(session=session)
  File "/Users/camilla.montonen/backups/development/recommender-pipelines/env/lib/python3.10/site-packages/airflow/utils/db.py", line 689, in _create_db_from_orm
    from airflow.www.session import AirflowDatabaseSessionInterface
  File "/Users/camilla.montonen/backups/development/recommender-pipelines/env/lib/python3.10/site-packages/airflow/www/session.py", line 21, in <module>
    from flask_session.sessions import SqlAlchemySessionInterface
ModuleNotFoundError: No module named 'flask_session.sessions'

It's probably some sort of version mismatch in my environment and flask_session.sessions has been deprecated or moved?
For reference, here are the verison numbers of the AirFlow dependencies in my environment:

(env) camilla.montonen@STL-D14J5637T2 pipelines % pip freeze | grep airflow
apache-airflow==2.6.3
apache-airflow-providers-cncf-kubernetes==8.1.1
apache-airflow-providers-common-sql==1.12.0
apache-airflow-providers-ftp==3.8.0
apache-airflow-providers-google==10.17.0
apache-airflow-providers-http==4.10.1
apache-airflow-providers-imap==3.5.0
apache-airflow-providers-sqlite==3.7.1
google-cloud-orchestration-airflow==1.12.1

Here are the version numbers of different Flask dependencies in this environment:

Flask==2.2.5
Flask-AppBuilder==4.3.1
Flask-Babel==2.0.0
Flask-Caching==2.2.0
Flask-JWT-Extended==4.6.0
Flask-Limiter==3.6.0
Flask-Login==0.6.3
Flask-Session==0.8.0
Flask-SQLAlchemy==2.5.1

I tried downgrading Flask-Session to both 0.6.0 and 0.7.0 and running airflow db init again, but none of them worked.
I went to the flask-sessions source code and it looks like the file flask_sessions.sessions was deleted three months ago pallets-eco/flask-session@2bc7df1#diff-05f62e791256751941798c00cb59cc8cef2632d0e3f2e2c28b979d87259fc666 , but that same line in apache-airflow has not been changed for two years.

I installed airflow directly from source and played around a little bit - turns out the required Flask-Session version is 0.5.0 - not sure why dependency resolution went wrong there, but anyway that solved it.

Installing airflow directly from source also appears to have added a new option to run airflow db migrate.
This however fails again with this message:

(env) camilla.montonen@STL-D14J5637T2 pipelines % airflow db migrate
DB: sqlite:////Users/camilla.montonen/airflow/airflow.db
Performing upgrade to the metadata database sqlite:////Users/camilla.montonen/airflow/airflow.db
[2024-05-05T13:28:50.915+0200] {migration.py:216} INFO - Context impl SQLiteImpl.
[2024-05-05T13:28:50.917+0200] {migration.py:219} INFO - Will assume non-transactional DDL.
[2024-05-05T13:28:50.918+0200] {db.py:1626} INFO - Creating tables
INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
ERROR [airflow.models.dagbag.DagBag] Failed to import: /Users/camilla.montonen/backups/development/recommender-pipelines/env/lib/python3.10/site-packages/airflow/example_dags/example_subdag_operator.py
Traceback (most recent call last):
  File "/Users/camilla.montonen/backups/development/recommender-pipelines/env/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1910, in _execute_context
    self.dialect.do_execute(
  File "/Users/camilla.montonen/backups/development/recommender-pipelines/env/lib/python3.10/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute
    cursor.execute(statement, parameters)
sqlite3.OperationalError: no such column: slot_pool.include_deferred

For anyone else wondering, this is how I solved it:

1. First run airflow db reset
2. Then airflow db migrate

[Taragolis]

It looks like there is no migrate option under the db commands:

It was added in Airflow 2.7

ModuleNotFoundError: No module named 'flask_session.sessions'

Seems like you've install incompatible version of Flask-Session
It is recommend to follow Installation from PyPI: Constraints, especially for old releases, otherwise you might be surprised when install incompatible/not-tested dependencies.

[camilla-storytel]

Seems like you've install incompatible version of Flask-Session

Perhaps this is what happened. I did start from a clean Python virtualenv, though and only installed airflow and some providers. I will see if I can replicate.