Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/households/controller_webhooks.py: 100%

1from datetime import UTC, datetime 

2from functools import cached_property 

3 

4from fastapi import APIRouter, BackgroundTasks, Depends 

5from pydantic import UUID4 

6 

7from mealie.routes._base.base_controllers import BaseUserController 

8from mealie.routes._base.controller import controller 

9from mealie.routes._base.mixins import HttpRepo 

10from mealie.schema import mapper 

11from mealie.schema.household.webhook import CreateWebhook, ReadWebhook, SaveWebhook, WebhookPagination 

12from mealie.schema.response.pagination import PaginationQuery 

13from mealie.services.scheduler.tasks.post_webhooks import post_group_webhooks, post_test_webhook 

14 

15router = APIRouter(prefix="/households/webhooks", tags=["Households: Webhooks"]) 

16 

17 

18@controller(router) 

19class ReadWebhookController(BaseUserController): 

20 @cached_property 

21 def repo(self): 

22 return self.repos.webhooks 

23 

24 @property 

25 def mixins(self) -> HttpRepo: 

26 return HttpRepo[CreateWebhook, SaveWebhook, CreateWebhook](self.repo, self.logger) 

27 

28 @router.get("", response_model=WebhookPagination) 

29 def get_all(self, q: PaginationQuery = Depends(PaginationQuery)): 

30 response = self.repo.page_all( 

31 pagination=q, 

32 override=ReadWebhook, 

33 ) 

34 

35 response.set_pagination_guides(router.url_path_for("get_all"), q.model_dump()) 

36 return response 

37 

38 @router.post("", response_model=ReadWebhook, status_code=201) 

39 def create_one(self, data: CreateWebhook): 

40 save = mapper.cast(data, SaveWebhook, group_id=self.group_id, household_id=self.household_id) 

41 return self.mixins.create_one(save) 

42 

43 @router.post("/rerun") 

44 def rerun_webhooks(self): 

45 """Manually re-fires all previously scheduled webhooks for today""" 

46 

47 start_time = datetime.min.time() 

48 start_dt = datetime.combine(datetime.now(UTC).date(), start_time) 

49 post_group_webhooks(start_dt=start_dt, group_id=self.group.id, household_id=self.household.id) 

50 

51 @router.get("/{item_id}", response_model=ReadWebhook) 

52 def get_one(self, item_id: UUID4): 

53 return self.mixins.get_one(item_id) 

54 

55 @router.post("/{item_id}/test") 

56 def test_one(self, item_id: UUID4, bg_tasks: BackgroundTasks): 

57 webhook = self.mixins.get_one(item_id) 

58 bg_tasks.add_task(post_test_webhook, webhook, "Test Webhook") 

59 

60 @router.put("/{item_id}", response_model=ReadWebhook) 

61 def update_one(self, item_id: UUID4, data: CreateWebhook): 

62 return self.mixins.update_one(data, item_id) 

63 

64 @router.delete("/{item_id}", response_model=ReadWebhook) 

65 def delete_one(self, item_id: UUID4): 

66 return self.mixins.delete_one(item_id) # type: ignore