Coverage for /usr/local/lib/python3.12/site-packages/prefect/_experimental/sla/client.py: 26%

1from __future__ import annotations 1 ctx1a

2 

3from typing import TYPE_CHECKING 1 ctx1a

4 

5from prefect.client.orchestration.base import BaseAsyncClient, BaseClient 1 ctx1a

6 

7if TYPE_CHECKING: 7 ↛ 8line 7 didn't jump to line 8 because the condition on line 7 was never true1 ctx1a

8 from uuid import UUID 

9 

10 from prefect._experimental.sla.objects import SlaMergeResponse, SlaTypes 

11 

12 

13class SlaClient(BaseClient): 1 ctx1a

14 def apply_slas_for_deployment( 1 ctx1a

15 self, deployment_id: "UUID", slas: "list[SlaTypes]" 

16 ) -> "SlaMergeResponse": 

17 """ 

18 Applies service level agreements for a deployment. Performs matching by SLA name. If a SLA with the same name already exists, it will be updated. If a SLA with the same name does not exist, it will be created. Existing SLAs that are not in the list will be deleted. 

19 Args: 

20 deployment_id: The ID of the deployment to update SLAs for 

21 slas: List of SLAs to associate with the deployment 

22 Raises: 

23 httpx.RequestError: if the SLAs were not updated for any reason 

24 Returns: 

25 SlaMergeResponse: The response from the backend, containing the names of the created, updated, and deleted SLAs 

26 """ 

27 resource_id = f"prefect.deployment.{deployment_id}" 

28 

29 for sla in slas: 

30 sla.set_deployment_id(deployment_id) 

31 

32 slas_spec_list = [ 

33 sla.model_dump(mode="json", exclude_unset=True) for sla in slas 

34 ] 

35 

36 response = self.request( 

37 "POST", 

38 f"/slas/apply-resource-slas/{resource_id}", 

39 json=slas_spec_list, 

40 ) 

41 response.raise_for_status() 

42 

43 response_json = response.json() 

44 

45 from prefect._experimental.sla.objects import SlaMergeResponse 

46 

47 return SlaMergeResponse( 

48 created=[sla.get("name") for sla in response_json.get("created")], 

49 updated=[sla.get("name") for sla in response_json.get("updated")], 

50 deleted=[sla.get("name") for sla in response_json.get("deleted")], 

51 ) 

52 

53 

54class SlaAsyncClient(BaseAsyncClient): 1 ctx1a

55 async def apply_slas_for_deployment( 1 ctx1a

56 self, deployment_id: "UUID", slas: "list[SlaTypes]" 

57 ) -> "UUID": 

58 """ 

59 Applies service level agreements for a deployment. Performs matching by SLA name. If a SLA with the same name already exists, it will be updated. If a SLA with the same name does not exist, it will be created. Existing SLAs that are not in the list will be deleted. 

60 Args: 

61 deployment_id: The ID of the deployment to update SLAs for 

62 slas: List of SLAs to associate with the deployment 

63 Raises: 

64 httpx.RequestError: if the SLAs were not updated for any reason 

65 Returns: 

66 SlaMergeResponse: The response from the backend, containing the names of the created, updated, and deleted SLAs 

67 """ 

68 resource_id = f"prefect.deployment.{deployment_id}" 

69 

70 for sla in slas: 

71 sla.set_deployment_id(deployment_id) 

72 

73 slas_spec_list = [ 

74 sla.model_dump(mode="json", exclude_unset=True) for sla in slas 

75 ] 

76 

77 response = await self.request( 

78 "POST", 

79 f"/slas/apply-resource-slas/{resource_id}", 

80 json=slas_spec_list, 

81 ) 

82 response.raise_for_status() 

83 

84 response_json = response.json() 

85 

86 from prefect._experimental.sla.objects import SlaMergeResponse 

87 

88 return SlaMergeResponse( 

89 created=[sla.get("name") for sla in response_json.get("created")], 

90 updated=[sla.get("name") for sla in response_json.get("updated")], 

91 deleted=[sla.get("name") for sla in response_json.get("deleted")], 

92 )