Coverage for /home/airflow/.local/lib/python3.12/site-packages/airflow/api_fastapi/core_api/routes/public/dag_warning.py: 100%

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 Annotated 

21 

22from fastapi import Depends 

23from sqlalchemy import select 

24from sqlalchemy.orm import joinedload 

25 

26from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity 

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

28 SessionDep, 

29 paginated_select, 

30) 

31from airflow.api_fastapi.common.parameters import ( 

32 FilterParam, 

33 QueryLimit, 

34 QueryOffset, 

35 SortParam, 

36 filter_param_factory, 

37) 

38from airflow.api_fastapi.common.router import AirflowRouter 

39from airflow.api_fastapi.core_api.datamodels.dag_warning import ( 

40 DAGWarningCollectionResponse, 

41) 

42from airflow.api_fastapi.core_api.security import ReadableDagWarningsFilterDep, requires_access_dag 

43from airflow.models.dagwarning import DagWarning, DagWarningType 

44 

45dag_warning_router = AirflowRouter(tags=["DagWarning"]) 

46 

47 

48@dag_warning_router.get( 

49 "/dagWarnings", 

50 dependencies=[Depends(requires_access_dag(method="GET", access_entity=DagAccessEntity.WARNING))], 

51) 

52def list_dag_warnings( 

53 dag_id: Annotated[FilterParam[str | None], Depends(filter_param_factory(DagWarning.dag_id, str | None))], 

54 warning_type: Annotated[ 

55 FilterParam[DagWarningType | None], 

56 Depends(filter_param_factory(DagWarning.warning_type, DagWarningType | None)), 

57 ], 

58 limit: QueryLimit, 

59 offset: QueryOffset, 

60 order_by: Annotated[ 

61 SortParam, 

62 Depends(SortParam(["dag_id", "warning_type", "message", "timestamp"], DagWarning).dynamic_depends()), 

63 ], 

64 readable_dag_warning_filter: ReadableDagWarningsFilterDep, 

65 session: SessionDep, 

66) -> DAGWarningCollectionResponse: 

67 """Get a list of Dag warnings.""" 

68 dag_warnings_select, total_entries = paginated_select( 

69 statement=select(DagWarning).options(joinedload(DagWarning.dag_model)), 

70 filters=[warning_type, dag_id, readable_dag_warning_filter], 

71 order_by=order_by, 

72 offset=offset, 

73 limit=limit, 

74 session=session, 

75 ) 

76 dag_warnings = session.scalars(dag_warnings_select) 

77 

78 return DAGWarningCollectionResponse( 

79 dag_warnings=dag_warnings, 

80 total_entries=total_entries, 

81 )