Coverage for api/endpoints/orders/bids.py: 58%

1from typing import Annotated 

2 

3from exceptions import NoAccessHTTPError 

4from fastapi import APIRouter, Depends 

5from handlers.authorization.check_role import has_role 

6from handlers.authorization.company_employee_access import ( 

7 flexible_company_assertion_query, 

8) 

9from handlers.grabbers.biding import bids_data_grabber 

10from handlers.orders.bids import on_bid_deletion, on_get_bids 

11from keycloak import idp 

12from mongodb import orders_col 

13from sotrans_models.models._mongo import PydanticObjectIdPath 

14from sotrans_models.models.orders.bid import BidDBModel 

15from sotrans_models.models.responses import ErrorRepr, GenericGetListResponse 

16from sotrans_models.models.roles import SotransRole 

17from sotrans_models.models.users import SotransOIDCUserModel 

18from starlette import status 

19from utils.access_wrapper import get_active_user 

20from utils.data_grabber import BaseGetListQueryParams, BaseGetOneQueryParams 

21from utils.helper import get_org_oid 

22 

23bids_router = APIRouter(prefix="/bids", tags=["bids"]) 

24 

25 

26@bids_router.delete( 

27 "/{bid_id}", 

28 responses={ 

29 204: {"description": "No content"}, 

30 404: {"model": ErrorRepr}, 

31 406: {"model": ErrorRepr}, 

32 }, 

33 status_code=status.HTTP_204_NO_CONTENT, 

34) 

35async def take_bid_away( 

36 bid_id: PydanticObjectIdPath, 

37 user=Depends(get_active_user([SotransRole.carrier_logistician])), 

38): 

39 await on_bid_deletion(user, bid_id) 

40 

41 

42@bids_router.get("", response_model_exclude={"text_search"}) 

43async def get_bids( 

44 user: Annotated[ 

45 SotransOIDCUserModel, 

46 Depends( 

47 idp.get_current_user( 

48 required_role_names=[SotransRole.company_logistician] 

49 ) 

50 ), 

51 ], 

52 params: BaseGetListQueryParams = Depends(), 

53) -> GenericGetListResponse[BidDBModel]: 

54 return await on_get_bids(user, params) 

55 

56 

57@bids_router.get( 

58 "/{bid_id}", 

59 responses={404: {"model": ErrorRepr}}, 

60 response_model_exclude={"text_search"}, 

61) 

62async def get_bid_by_id( 

63 bid_id: PydanticObjectIdPath, 

64 user=Depends(get_active_user([SotransRole.carrier_logistician])), 

65 params: BaseGetOneQueryParams = Depends(), 

66) -> BidDBModel: 

67 if has_role(user, SotransRole.company_director): 

68 return await bids_data_grabber.get_one(bid_id, params) 

69 

70 if has_role(user, SotransRole.company_logistician): 

71 

72 async def raise_on_not_assigned(bid: BidDBModel): 

73 company_q = await flexible_company_assertion_query(user) 

74 if ( 

75 await orders_col.collection.find_one( 

76 {"id": bid.order_id} | company_q 

77 ) 

78 is None 

79 ): 

80 raise NoAccessHTTPError("заказ") 

81 

82 return await bids_data_grabber.get_one_by_id_with_pattern( 

83 bid_id, params, processors=[raise_on_not_assigned] 

84 ) 

85 

86 organization_id = get_org_oid(user) 

87 return await bids_data_grabber.get_one_by_id_with_pattern( 

88 bid_id, params, {BidDBModel.carrier.id: organization_id} 

89 )