Coverage for /home/airflow/.local/lib/python3.12/site-packages/airflow/api_fastapi/common/db/dags.py: 83%

1# Licensed to the Apache Software Foundation (ASF) under one 

2# or more contributor license agreements. See the NOTICE file 

3# distributed with this work for additional information 

4# regarding copyright ownership. The ASF licenses this file 

5# to you under the Apache License, Version 2.0 (the 

6# "License"); you may not use this file except in compliance 

7# with the License. You may obtain a copy of the License at 

8# 

9# http://www.apache.org/licenses/LICENSE-2.0 

10# 

11# Unless required by applicable law or agreed to in writing, 

12# software distributed under the License is distributed on an 

13# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 

14# KIND, either express or implied. See the License for the 

15# specific language governing permissions and limitations 

16# under the License. 

17 

18from __future__ import annotations 

19 

20from typing import TYPE_CHECKING 

21 

22from sqlalchemy import func, select 

23from sqlalchemy.orm import selectinload 

24 

25from airflow.api_fastapi.common.db.common import ( 

26 apply_filters_to_select, 

27) 

28from airflow.api_fastapi.common.parameters import BaseParam, RangeFilter, SortParam 

29from airflow.models import DagModel 

30from airflow.models.dagrun import DagRun 

31 

32if TYPE_CHECKING: 32 ↛ 33line 32 didn't jump to line 33 because the condition on line 32 was never true

33 from sqlalchemy.sql import Select 

34 

35 

36def generate_dag_with_latest_run_query( 

37 max_run_filters: list[BaseParam], order_by: SortParam, *, dag_ids: set[str] | None = None 

38) -> Select: 

39 """ 

40 Generate a query to fetch Dags with their latest run. 

41 

42 :param max_run_filters: List of filters to apply to the latest run 

43 :param order_by: Sort parameter for ordering results 

44 :param dag_ids: Optional set of Dag IDs to limit the query to. When provided, both the main 

45 Dag query and the subquery for finding the latest runs will be filtered to 

46 only these Dag IDs, improving performance when users have limited Dag access. 

47 :return: SQLAlchemy Select statement 

48 """ 

49 query = select(DagModel).options(selectinload(DagModel.tags)) 

50 

51 # Filter main query by dag_ids if provided 

52 if dag_ids is not None: 52 ↛ 56line 52 didn't jump to line 56 because the condition on line 52 was always true

53 query = query.where(DagModel.dag_id.in_(dag_ids or set())) 

54 

55 # Also filter the subquery for finding latest runs 

56 max_run_id_query_stmt = select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id")) 

57 if dag_ids is not None: 57 ↛ 59line 57 didn't jump to line 59 because the condition on line 57 was always true

58 max_run_id_query_stmt = max_run_id_query_stmt.where(DagRun.dag_id.in_(dag_ids or set())) 

59 max_run_id_query = max_run_id_query_stmt.group_by(DagRun.dag_id).subquery(name="mrq") 

60 

61 has_max_run_filter = False 

62 

63 for max_run_filter in max_run_filters: 63 ↛ 72line 63 didn't jump to line 72 because the loop on line 63 didn't complete

64 if isinstance(max_run_filter, RangeFilter): 64 ↛ 68line 64 didn't jump to line 68 because the condition on line 64 was always true

65 if max_run_filter.is_active(): 

66 has_max_run_filter = True 

67 break 

68 if max_run_filter.value: 68 ↛ 63line 68 didn't jump to line 63 because the condition on line 68 was always true

69 has_max_run_filter = True 

70 break 

71 

72 requested_order_by_set = set(order_by.value) if order_by.value is not None else set() 

73 dag_run_order_by_set = set( 

74 ["last_run_state", "last_run_start_date", "-last_run_state", "-last_run_start_date"], 

75 ) 

76 

77 if has_max_run_filter or (requested_order_by_set & dag_run_order_by_set): 77 ↛ 84line 77 didn't jump to line 84 because the condition on line 77 was always true

78 query = query.join( 

79 max_run_id_query, 

80 DagModel.dag_id == max_run_id_query.c.dag_id, 

81 isouter=True, 

82 ).join(DagRun, DagRun.id == max_run_id_query.c.max_dag_run_id, isouter=True) 

83 

84 if has_max_run_filter: 84 ↛ 90line 84 didn't jump to line 90 because the condition on line 84 was always true

85 query = apply_filters_to_select( 

86 statement=query, 

87 filters=max_run_filters, 

88 ) 

89 

90 return query