Coverage for /home/airflow/.local/lib/python3.12/site-packages/airflow/api_fastapi/core_api/routes/ui/backfills.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. 

17from __future__ import annotations 

18 

19from typing import Annotated 

20 

21from fastapi import Depends, status 

22from sqlalchemy import select 

23from sqlalchemy.orm import joinedload 

24 

25from airflow.api_fastapi.common.db.common import SessionDep, paginated_select 

26from airflow.api_fastapi.common.parameters import ( 

27 FilterOptionEnum, 

28 FilterParam, 

29 QueryLimit, 

30 QueryOffset, 

31 SortParam, 

32 filter_param_factory, 

33) 

34from airflow.api_fastapi.common.router import AirflowRouter 

35from airflow.api_fastapi.core_api.datamodels.backfills import BackfillCollectionResponse, BackfillResponse 

36from airflow.api_fastapi.core_api.openapi.exceptions import ( 

37 create_openapi_http_exception_doc, 

38) 

39from airflow.api_fastapi.core_api.security import ReadableBackfillsFilterDep, requires_access_backfill 

40from airflow.models.backfill import Backfill 

41 

42backfills_router = AirflowRouter(tags=["Backfill"], prefix="/backfills") 

43 

44 

45@backfills_router.get( 

46 path="", 

47 responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]), 

48 dependencies=[ 

49 Depends(requires_access_backfill(method="GET")), 

50 ], 

51) 

52def list_backfills_ui( 

53 limit: QueryLimit, 

54 offset: QueryOffset, 

55 order_by: Annotated[ 

56 SortParam, 

57 Depends(SortParam(["id"], Backfill).dynamic_depends()), 

58 ], 

59 readable_backfills_filter: ReadableBackfillsFilterDep, 

60 session: SessionDep, 

61 dag_id: Annotated[FilterParam[str | None], Depends(filter_param_factory(Backfill.dag_id, str | None))], 

62 active: Annotated[ 

63 FilterParam[bool | None], 

64 Depends(filter_param_factory(Backfill.completed_at, bool | None, FilterOptionEnum.IS_NONE, "active")), 

65 ], 

66) -> BackfillCollectionResponse: 

67 select_stmt, total_entries = paginated_select( 

68 statement=select(Backfill).options(joinedload(Backfill.dag_model)), 

69 filters=[dag_id, active, readable_backfills_filter], 

70 order_by=order_by, 

71 offset=offset, 

72 limit=limit, 

73 session=session, 

74 ) 

75 backfills = [ 

76 BackfillResponse(**row._mapping) if not isinstance(row, Backfill) else row 

77 for row in session.scalars(select_stmt) 

78 ] 

79 return BackfillCollectionResponse( 

80 backfills=backfills, 

81 total_entries=total_entries, 

82 )