Coverage for mpcforces_extractor\api\routes\nodes.py: 29%

1from typing import List 

2from fastapi import APIRouter, Depends, HTTPException, Query 

3from mpcforces_extractor.api.db.database import NodeDBModel 

4from mpcforces_extractor.api.dependencies import get_db 

5from mpcforces_extractor.api.config import ITEMS_PER_PAGE 

6 

7router = APIRouter() 

8 

9 

10@router.get("", response_model=List[NodeDBModel]) 

11async def get_nodes( 

12 page: int = Query(1, ge=1), db=Depends(get_db) 

13) -> List[NodeDBModel]: 

14 """ 

15 Get nodes with pagination (fixed 100 items per page) 

16 """ 

17 # Calculate offset based on the current page 

18 offset = (page - 1) * ITEMS_PER_PAGE 

19 

20 # Fetch nodes from the database with the calculated offset and limit (fixed at 100) 

21 nodes = await db.get_nodes(offset=offset, limit=ITEMS_PER_PAGE) 

22 

23 # Handle case when no nodes are found 

24 if not nodes: 

25 raise HTTPException(status_code=404, detail="No nodes found") 

26 

27 return nodes 

28 

29 

30@router.get("/all", response_model=List[NodeDBModel]) 

31async def get_all_nodes(db=Depends(get_db)) -> int: 

32 """ 

33 Get all nodes 

34 """ 

35 

36 nodes = await db.get_all_nodes() 

37 

38 if not nodes: 

39 raise HTTPException(status_code=404, detail="No nodes found") 

40 

41 return nodes 

42 

43 

44@router.get("/filter/{filter_input}", response_model=List[NodeDBModel]) 

45async def get_nodes_filtered( 

46 filter_input: str, db=Depends(get_db) 

47) -> List[NodeDBModel]: 

48 """ 

49 Get nodes filtered by a string, get it from all nodes, not paginated. 

50 The filter can be a range like '1-3' or comma-separated values like '1,2,3'. 

51 """ 

52 nodes = await db.get_all_nodes() 

53 filtered_nodes = [] 

54 

55 # Split the filter string by comma and process each part 

56 filter_parts = filter_input.split(",") 

57 for part in filter_parts: 

58 part = part.strip() # Trim whitespace 

59 if "-" in part: 

60 # Handle range like '1-3' 

61 start, end = part.split("-") 

62 try: 

63 start_id = int(start.strip()) 

64 end_id = int(end.strip()) 

65 filtered_nodes.extend( 

66 node for node in nodes if start_id <= node.id <= end_id 

67 ) 

68 except ValueError: 

69 raise HTTPException( 

70 status_code=400, detail="Invalid range in filter" 

71 ) from ValueError 

72 else: 

73 # Handle single ID 

74 try: 

75 node_id = int(part) 

76 node = next((node for node in nodes if node.id == node_id), None) 

77 if node: 

78 filtered_nodes.append(node) 

79 except ValueError: 

80 raise HTTPException( 

81 status_code=400, detail="Invalid ID in filter" 

82 ) from ValueError 

83 return filtered_nodes