Coverage for api/handlers/orders/archive.py: 22%

1import datetime 

2import uuid 

3from enum import Enum 

4from typing import Any 

5 

6import aiofiles 

7import dicttoxml 

8from bson import ObjectId 

9from handlers.authorization.check_role import has_role 

10from handlers.authorization.confidential import hide_model_fields_from_carrier 

11from handlers.grabbers.orders import archive_data_grabber 

12from mongodb import archive_col 

13from sotrans_models.models.orders.order import OrderDBModel 

14from sotrans_models.models.responses import ( 

15 GenericGetListResponse, 

16 MonthlyStatisticsResponse, 

17) 

18from sotrans_models.models.roles import SotransRole 

19from sotrans_models.models.users import SotransOIDCUserModel 

20from utils.data_grabber import BaseGetListQueryParams, update_search_query 

21from utils.dt_utils import get_current_datetime, months_delta 

22 

23 

24async def get_archive_orders( 

25 user: SotransOIDCUserModel, params: BaseGetListQueryParams 

26) -> GenericGetListResponse[OrderDBModel]: 

27 is_carrier = not ( 

28 has_role(user, SotransRole.drivers_bot) 

29 or has_role(user, SotransRole.company_logistician) 

30 ) 

31 if is_carrier: 

32 carriers_restriction_query = { 

33 OrderDBModel.carrier.id: user.organization_id 

34 } 

35 params.where = update_search_query( 

36 params.where, carriers_restriction_query 

37 ) 

38 orders_get_list = await archive_data_grabber.get_list(params, user) 

39 if is_carrier: 

40 for order in orders_get_list.items: 

41 hide_model_fields_from_carrier(order) 

42 return orders_get_list 

43 

44 

45def check_enums(model_dump: dict[str, Any]): 

46 for k in model_dump: 

47 v = model_dump[k] 

48 if isinstance(v, (ObjectId, uuid.UUID)): 

49 model_dump[k] = str(v) 

50 if isinstance(v, datetime.datetime): 

51 model_dump[k] = v.strftime("%d.%m.%Y %H:%M:%S") 

52 if isinstance(v, Enum): 

53 model_dump[k] = v.value 

54 elif isinstance(v, dict): 

55 check_enums(v) 

56 elif isinstance(v, list): 

57 model_dump[k] = [i.value if isinstance(i, Enum) else i for i in v] 

58 for i in v: 

59 if isinstance(i, dict): 

60 check_enums(i) 

61 

62 

63async def export_to_xml(orders: list[dict[str, Any] | OrderDBModel]) -> str: 

64 xmlable = [] 

65 for o in orders: 

66 if isinstance(o, OrderDBModel): 

67 xmlable.append(o.model_dump(format_ids=False)) 

68 else: 

69 xmlable.append(o) 

70 data_to_xml = {"orders": xmlable} 

71 check_enums(data_to_xml) 

72 xml = dicttoxml.dicttoxml(data_to_xml) 

73 workdir_path = "./assets" 

74 xml_export_file_name = f"{workdir_path}/{uuid.uuid4()}.xml" 

75 async with aiofiles.open(xml_export_file_name, "wb") as xef: 

76 await xef.write(xml) 

77 return xml_export_file_name 

78 

79 

80async def on_get_archive_stats(): 

81 current_dt = get_current_datetime() 

82 stats_data = {"graph": []} 

83 for i in range(1, 13): 

84 start_point = months_delta(-i, current_dt) 

85 end_point = months_delta(-i + 1, current_dt) 

86 archive_per_month = await archive_col.count( 

87 { 

88 OrderDBModel.created_at: { 

89 "$and": [{"$lt": end_point}, {"$gte": start_point}] 

90 } 

91 } 

92 ) 

93 stats_data["graph"].append({"key": i, "value": archive_per_month}) 

94 stats_data["total"] = await archive_col.count({}) 

95 return MonthlyStatisticsResponse(**stats_data)