Coverage for src \ ingress_recipient \ fastapi.py: 100%

1"""FastAPI dependency for validating recipient data.""" 

2 

3from typing import Annotated 

4 

5from fastapi import Depends 

6 

7from .types import RecipientData 

8 

9 

10async def validate_recipient_data(data: RecipientData) -> RecipientData: 

11 """FastAPI dependency for validating recipient data. 

12 

13 Pydantic automatically validates the request body and returns 422 

14 for validation errors. This dependency exists for explicit clarity 

15 and to allow custom logic if needed in the future. 

16 

17 Args: 

18 data: Validated RecipientData from request body (auto-validated by Pydantic) 

19 

20 Returns: 

21 RecipientData object 

22 

23 Raises: 

24 HTTPException: 422 if validation fails (automatic via Pydantic) 

25 

26 Example: 

27 ```python 

28 from ingress_recipient import RecipientData, validate_recipient_data 

29 from fastapi import APIRouter, Depends 

30 

31 router = APIRouter() 

32 

33 @router.post("/destinations/telemetry") 

34 async def receive_telemetry( 

35 data: RecipientData = Depends(validate_recipient_data), 

36 ): 

37 # data is fully typed and validated 

38 print(f"Received data from channel {data.channel_id}") 

39 

40 # Process data... 

41 await save_telemetry(data) 

42 

43 return {"status": "accepted"} 

44 ``` 

45 

46 You can also use the ValidatedRecipientData type alias for cleaner code: 

47 

48 ```python 

49 from ingress_recipient import ValidatedRecipientData 

50 

51 @router.post("/destinations/telemetry") 

52 async def receive_telemetry(data: ValidatedRecipientData): 

53 # data is fully typed and validated 

54 print(f"Received data from channel {data.channel_id}") 

55 return {"status": "accepted"} 

56 ``` 

57 """ 

58 return data 

59 

60 

61# Type alias for cleaner endpoint signatures 

62ValidatedRecipientData = Annotated[RecipientData, Depends(validate_recipient_data)] 

63"""Type alias for validated recipient data in FastAPI endpoints. 

64 

65This combines the RecipientData type with the validation dependency, 

66allowing for cleaner endpoint signatures. 

67 

68Example: 

69 ```python 

70 from ingress_recipient import ValidatedRecipientData 

71 

72 @router.post("/destinations/telemetry") 

73 async def receive_telemetry(data: ValidatedRecipientData): 

74 # data is automatically validated 

75 print(f"Channel: {data.channel_id}") 

76 return {"status": "accepted"} 

77 ``` 

78"""