[AIRFLOW-6856] BugFix: Paused Dags still Scheduled

[kaxil]

#7476 introduced a bug due to which Paused Dags where still Scheduled.

The Bug is the following query returns a list of sets:

Query:

paused_dag_ids = (
            session.query(DagModel.dag_id)
            .filter(DagModel.is_paused.is_(True))
            .filter(DagModel.dag_id.in_(dagbag.dag_ids))
            .all()
        )

Result:

[('example_bash_operator',)]

Hence in _find_dags_to_process() (below):

        if len(self.dag_ids) > 0:
            dags = [dag for dag in dags
                    if dag.dag_id in self.dag_ids and
                    dag.dag_id not in paused_dag_ids]
        else:
            dags = [dag for dag in dags
                    if dag.dag_id not in paused_dag_ids]
        return dags

following happens:

dags = [dag for dag in dags if "example_bash_operator" not in [('example_bash_operator',)] ] 

This evaluates to false. Instead paused_dag_ids should be {'example_bash_operator'} (A set) or just a list ['example_bash_operator']

Simplified Problem and solution:

In [1]: a = [('example_bash_operator',)]

In [2]: b = 'example_bash_operator'

In [3]: set(aa for aa in a)
Out[3]: {('example_bash_operator',)}

In [4]: b in a
Out[4]: False

In [5]: set(aa for aa, in a)
Out[5]: {'example_bash_operator'}

In [6]: b in set(aa for aa, in a)
Out[6]: True

---

Issue link: AIRFLOW-6856

Make sure to mark the boxes below before creating PR: [x]

• Description above provides context of the change
• Commit message/PR title starts with [AIRFLOW-NNNN]. AIRFLOW-NNNN = JIRA ID^*
• Unit tests coverage for changes (not needed for documentation changes)
• Commits follow "How to write a good git commit message"
• Relevant documentation is updated including usage instructions.
• I will engage committers as explained in Contribution Workflow Example.

^* For document-only changes commit message can start with [AIRFLOW-XXXX].

---

In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.
Read the Pull Request Guidelines for more information.

[zhongjiajie]

Found out this bug just now

[mik-laj]

Should we add tests to avoid regression?

[kaxil]

Should we add tests to avoid regression?

Created a new PR: #7587 for it