Coverage for api/endpoints/orders/external_orders.py: 50%

1from typing import Annotated, Any 

2 

3from fastapi import APIRouter, Depends 

4from handlers.orders.external_orders import ( 

5 on_delete_batch, 

6 on_patch_batch, 

7 on_post_batch, 

8) 

9from keycloak import idp 

10from mongodb import buf_col, orders_col, trash_orders_col, users_col 

11from services.notifications.director import ( 

12 ScrapedNotificationContainer, 

13 notification_api, 

14) 

15from sotrans_models.models.orders.order import ( 

16 ExternalOrderStatus, 

17 OrderDBModel, 

18) 

19from sotrans_models.models.parsers.scraped_api import ( 

20 ScrapedOrdersCreateModel, 

21 ScrapedOrdersModel, 

22) 

23from sotrans_models.models.roles import SotransRole 

24from sotrans_models.models.users import ( 

25 SotransOIDCUserModel, 

26 SotransUserDBModel, 

27) 

28from starlette import status 

29from stubs import scraped_container 

30 

31scraped_router = APIRouter( 

32 prefix="/external-orders", 

33 tags=["external-orders", "orders"], 

34 dependencies=[ 

35 Depends( 

36 idp.get_current_user(required_role_names=[SotransRole.scraper]) 

37 ) 

38 ], 

39) 

40 

41 

42@scraped_router.get("") 

43async def get_orders_batch( 

44 resource_name: str = None, show_missing: bool = False 

45) -> list[OrderDBModel]: 

46 if resource_name: 

47 pattern: dict[str, Any] = {"external.resource_name": resource_name} 

48 else: 

49 pattern = {"external.id": {"$ne": None}} 

50 if not show_missing: 

51 pattern["external.status"] = {"$ne": ExternalOrderStatus.missing.value} 

52 buf_orders = [ 

53 OrderDBModel(**order) 

54 async for order in buf_col.find_batch_raw(pattern) 

55 ] 

56 exchange_orders = [ 

57 OrderDBModel(**order) 

58 async for order in orders_col.find_batch_raw(pattern) 

59 ] 

60 buf_orders.extend(exchange_orders) 

61 trash_orders = [ 

62 OrderDBModel(**order) 

63 async for order in trash_orders_col.find_batch_raw(pattern) 

64 ] 

65 buf_orders.extend(trash_orders) 

66 return buf_orders 

67 

68 

69@scraped_router.post("", status_code=status.HTTP_201_CREATED) 

70async def save_orders_batch( 

71 orders: ScrapedOrdersCreateModel, 

72 user: Annotated[ 

73 SotransOIDCUserModel, 

74 Depends(idp.get_current_user()), 

75 ], 

76 notification_container: ScrapedNotificationContainer = Depends( 

77 scraped_container 

78 ), 

79): 

80 await on_post_batch(user, orders.items, notification_container) 

81 

82 

83@scraped_router.patch("") 

84async def patch_batch( 

85 orders: ScrapedOrdersCreateModel, 

86 _: Annotated[ 

87 SotransOIDCUserModel, 

88 Depends(idp.get_current_user()), 

89 ], 

90): 

91 """Updates by external_data->link.""" 

92 await on_patch_batch(orders.items) 

93 

94 

95@scraped_router.delete("", status_code=status.HTTP_204_NO_CONTENT) 

96async def delete_batch( 

97 orders: ScrapedOrdersModel, 

98 _: Annotated[ 

99 SotransOIDCUserModel, 

100 Depends(idp.get_current_user()), 

101 ], 

102): 

103 """Updates by external_data->link.""" 

104 await on_delete_batch(orders.items) 

105 

106 

107@scraped_router.post("/error") 

108async def on_error( 

109 error: dict, 

110 _: Annotated[ 

111 SotransOIDCUserModel, 

112 Depends(idp.get_current_user()), 

113 ], 

114): 

115 admins = await users_col.find_batch({"role": SotransRole.admin.value}) 

116 if admins: 

117 notification_api.admin_notification_on_parser( 

118 error, [SotransUserDBModel(**a) for a in admins] 

119 )