Coverage for api/endpoints/admin/ads.py: 50%

1from typing import Annotated 

2 

3from database.entity import update_etag 

4from exceptions import NoAccessHTTPError, NotFoundHTTPError 

5from fastapi import APIRouter, Depends 

6from handlers.authorization.check_role import has_role 

7from handlers.grabbers.administrative import ads_data_grabber 

8from keycloak import idp 

9from mongodb import ads_col 

10from pymongo import ReturnDocument 

11from pymongo.results import DeleteResult 

12from sotrans_models.models._mongo import PydanticObjectIdPath 

13from sotrans_models.models.admin.ads import ( 

14 AdvertisingBlockDBModel, 

15 CreateAdvertisingBlockModel, 

16 UpdateAdvertisingBlockModel, 

17) 

18from sotrans_models.models.responses import ErrorRepr, GenericGetListResponse 

19from sotrans_models.models.roles import SotransRole 

20from sotrans_models.models.users import SotransOIDCUserModel 

21from starlette import status 

22from starlette.responses import Response 

23from utils.data_grabber import ( 

24 BaseGetListQueryParams, 

25 BaseGetOneQueryParams, 

26 MongoDataGrabber, 

27) 

28from utils.helper import etag_detalizer, get_hash 

29 

30ads_router = APIRouter(prefix="/ads-block", tags=["advertising"]) 

31 

32 

33@ads_router.get("") 

34async def get_ads_blocks( 

35 user: Annotated[ 

36 SotransOIDCUserModel, 

37 Depends( 

38 idp.get_current_user( 

39 required_role_names=[SotransRole.carrier_logistician] 

40 ) 

41 ), 

42 ], 

43 response: Response, 

44 params: BaseGetListQueryParams = Depends(), 

45) -> GenericGetListResponse[AdvertisingBlockDBModel]: 

46 where = MongoDataGrabber.parse_where(params.where) 

47 if ( 

48 where is None 

49 or where.get(AdvertisingBlockDBModel.is_active) is not True 

50 ) and not has_role(user, SotransRole.admin): 

51 raise NoAccessHTTPError("неактивная реклама") 

52 if ( 

53 where is not None 

54 and where.get(AdvertisingBlockDBModel.is_active) is True 

55 ): 

56 response.headers["Cache-Control"] = "max-age=20" 

57 # response.headers["Date"] = datetime.now().isoformat() 

58 return await ads_data_grabber.get_list(params) 

59 

60 

61@ads_router.post("", status_code=status.HTTP_201_CREATED) 

62async def create_ads( 

63 ads: CreateAdvertisingBlockModel, 

64 _: Annotated[ 

65 SotransOIDCUserModel, 

66 Depends(idp.get_current_user(required_role_names=[SotransRole.admin])), 

67 ], 

68) -> AdvertisingBlockDBModel: 

69 ads_dump = ads.model_dump() 

70 etag = get_hash(ads_dump) 

71 return await ads_col.create({**ads_dump, **{"etag": etag}}) 

72 

73 

74@ads_router.delete("/{ads_id}", status_code=status.HTTP_204_NO_CONTENT) 

75async def remove_ads( 

76 ads_id: PydanticObjectIdPath, 

77 _: Annotated[ 

78 SotransOIDCUserModel, 

79 Depends(idp.get_current_user(required_role_names=[SotransRole.admin])), 

80 ], 

81 resource_with_etag: UpdateAdvertisingBlockModel | None = None, 

82): 

83 etag_q = ( 

84 {"etag": resource_with_etag.etag} 

85 if resource_with_etag is not None 

86 else {} 

87 ) 

88 deletion: DeleteResult = await ads_col.collection.delete_one( 

89 {"_id": ads_id} | etag_q 

90 ) 

91 if deletion.deleted_count == 0: 

92 await etag_detalizer(ads_col, etag_q, {"_id": ads_id}) 

93 raise NotFoundHTTPError("объявление с id и, возможно, etag") 

94 

95 

96@ads_router.patch("/{ads_id}", responses={404: {"model": ErrorRepr}}) 

97async def update_ads_block( 

98 ads_id: PydanticObjectIdPath, 

99 ads: UpdateAdvertisingBlockModel, 

100 _: Annotated[ 

101 SotransOIDCUserModel, 

102 Depends(idp.get_current_user(required_role_names=[SotransRole.admin])), 

103 ], 

104) -> AdvertisingBlockDBModel: 

105 etag_q = {"etag": ads.etag} if ads.etag else {} 

106 doc = await ads_col.collection.find_one_and_update( 

107 {"_id": ads_id} | etag_q, 

108 {"$set": ads.model_dump(exclude_none=True)}, 

109 return_document=ReturnDocument.AFTER, 

110 ) 

111 if not doc: 

112 await etag_detalizer(ads_col, etag_q, {"_id": ads_id}) 

113 raise NotFoundHTTPError("реклама") 

114 await update_etag(doc, ads_col) 

115 return doc 

116 

117 

118@ads_router.get("/{ads_id}", responses={404: {"model": ErrorRepr}}) 

119async def get_ads_by_id( 

120 ads_id: PydanticObjectIdPath, 

121 _: Annotated[ 

122 SotransOIDCUserModel, 

123 Depends(idp.get_current_user(required_role_names=[SotransRole.admin])), 

124 ], 

125 params: BaseGetOneQueryParams = Depends(), 

126) -> AdvertisingBlockDBModel: 

127 return await ads_data_grabber.get_one(ads_id, params)