Coverage for /home/airflow/.local/lib/python3.12/site-packages/airflow/api_fastapi/execution_api/datamodels/hitl.py: 94%

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 collections.abc import Iterable 

20from typing import Any 

21from uuid import UUID 

22 

23from pydantic import Field 

24 

25from airflow.api_fastapi.common.types import UtcDateTime 

26from airflow.api_fastapi.core_api.base import BaseModel 

27from airflow.models.hitl import HITLDetail 

28 

29 

30class HITLUser(BaseModel): 

31 """Schema for a Human-in-the-loop users.""" 

32 

33 id: str 

34 name: str 

35 

36 

37class HITLDetailRequest(BaseModel): 

38 """Schema for the request part of a Human-in-the-loop detail for a specific task instance.""" 

39 

40 ti_id: UUID 

41 options: list[str] = Field(min_length=1) 

42 subject: str 

43 body: str | None = None 

44 defaults: list[str] | None = None 

45 multiple: bool = False 

46 params: dict[str, Any] = Field(default_factory=dict) 

47 assigned_users: list[HITLUser] = Field(default_factory=list) 

48 

49 

50class UpdateHITLDetailPayload(BaseModel): 

51 """Schema for writing the response part of a Human-in-the-loop detail for a specific task instance.""" 

52 

53 ti_id: UUID 

54 chosen_options: list[str] = Field(min_length=1) 

55 params_input: dict[str, Any] = Field(default_factory=dict) 

56 

57 

58class HITLDetailResponse(BaseModel): 

59 """Schema for the response part of a Human-in-the-loop detail for a specific task instance.""" 

60 

61 response_received: bool 

62 responded_by_user: HITLUser | None = None 

63 responded_at: UtcDateTime | None 

64 # It's empty if the user has not yet responded. 

65 chosen_options: Iterable[str] | None 

66 params_input: dict[str, Any] = Field(default_factory=dict) 

67 

68 @classmethod 

69 def from_hitl_detail_orm(cls, hitl_detail: HITLDetail) -> HITLDetailResponse: 

70 hitl_user = ( 

71 HITLUser( 

72 id=hitl_detail.responded_by_user_id, 

73 name=hitl_detail.responded_by_user_name, 

74 ) 

75 if hitl_detail.responded_by_user 

76 else None 

77 ) 

78 

79 return HITLDetailResponse( 

80 response_received=hitl_detail.response_received, 

81 responded_at=hitl_detail.responded_at, 

82 responded_by_user=hitl_user, 

83 chosen_options=hitl_detail.chosen_options or (), 

84 params_input=hitl_detail.params_input or {}, 

85 )