Coverage for api/endpoints/int1c.py: 35%

1from typing import Any 

2 

3from fastapi import APIRouter, Depends 

4from handlers.authorization.by_api_key import check_access 

5from handlers.int1c import apply_processing_results, on_get_updates 

6from mongodb import clients_col, dbg_1c_col 

7from pymongo.results import UpdateResult 

8from sotrans_models.models.misc.client import ClientCreateModel, ClientDBModel 

9from sotrans_models.models.updates1c import ( 

10 Processing1CResult, 

11 Update1CModel, 

12 UpdateStatus, 

13) 

14from sotrans_models.utils.text_mappers import get_clients_text_search 

15 

16integration_router = APIRouter( 

17 prefix="/1c", dependencies=[Depends(check_access)], tags=["1c"] 

18) 

19 

20 

21@integration_router.post("/clients") 

22async def clients_1c(clients: list[ClientCreateModel]): 

23 clients_data: list[ClientDBModel] = [] 

24 for client in clients: 

25 client.responsible = None 

26 client.subsidiary = None 

27 if client.short_name is None: 

28 client.short_name = client.name 

29 cd = ClientDBModel(**client.model_dump()) 

30 cd.text_search = get_clients_text_search(cd) 

31 clients_data.append(cd) 

32 for cm in clients_data: 

33 update_res: UpdateResult = await clients_col.collection.update_one( 

34 {ClientDBModel.inn: cm.inn, ClientDBModel.kpp: cm.kpp}, 

35 {"$set": cm.model_dump(exclude_none=True)}, 

36 upsert=True, 

37 ) 

38 if update_res.modified_count == 0: 

39 await clients_col.update_by_id( 

40 update_res.upserted_id, {"id": update_res.upserted_id} 

41 ) 

42 

43 

44@integration_router.post("/vehicles") 

45async def make_vehicles(vehicles_data: list[dict[str, Any]]): 

46 # log_warning(str(vehicles_data)) 

47 # await write_1c_to_file(vehicles_data, "vehicles") 

48 await dbg_1c_col.create({"type": "vehicles", "data": vehicles_data}) 

49 for _ in vehicles_data: 

50 ... 

51 

52 

53@integration_router.post("/drivers") 

54async def make_drivers(drivers_data: list[dict[str, Any]]): 

55 await dbg_1c_col.create({"type": "drivers", "data": drivers_data}) 

56 for _ in drivers_data: 

57 ... 

58 

59 

60async def get_events( 

61 status_1c: UpdateStatus | None = None, 

62) -> list[Update1CModel]: 

63 return await on_get_updates(status_1c) 

64 

65 

66async def updates_statuses(updated: list[Processing1CResult]): 

67 await apply_processing_results(updated)