Coverage for rfpy/api/endpoints/categories.py: 100%

1""" 

2Use Categories to organize Projects 

3""" 

4 

5from typing import List 

6 

7from sqlalchemy.orm import Session 

8 

9from rfpy.suxint import http 

10from rfpy.web import serial 

11from rfpy.auth import perms 

12from rfpy.model import Category, User 

13from rfpy.model.exc import DuplicateDataProvided 

14from rfpy.api import fetch, validate 

15 

16 

17@http 

18def get_category(session: Session, user: User, category_id: int) -> serial.Category: 

19 """Fetch the Category object for the given ID""" 

20 return fetch.category_for_user(session, user, category_id).as_dict() 

21 

22 

23@http 

24def delete_category(session: Session, user: User, category_id: int): 

25 """ 

26 Delete the category with the given ID. 

27 

28 Note that projects belong to this category are __not__ deleted 

29 This action requires the user permission 'MANAGE_ORGANISATION' 

30 """ 

31 cat = fetch.category_for_user(session, user, category_id) 

32 validate.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

33 session.delete(cat) 

34 

35 

36@http 

37def post_category(session: Session, user: User, category_doc) -> serial.Category: 

38 """ 

39 Create a new Category belonging to the current user's organisation. 

40 

41 Categories with duplicate names are not permitted and receive an HTTP 409 Response 

42 

43 @permission MANAGE_CATEGORIES 

44 """ 

45 validate.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

46 cat_name = category_doc.name 

47 existing = ( 

48 session.query(Category) 

49 .filter(Category.organisation == user.organisation) 

50 .filter(Category.name == cat_name) 

51 .count() 

52 ) 

53 

54 if existing: 

55 raise DuplicateDataProvided(f"Category with name '{cat_name}' already exists") 

56 

57 cat = Category( 

58 name=cat_name, 

59 description=category_doc.description, 

60 organisation=user.organisation, 

61 ) 

62 

63 session.add(cat) 

64 session.flush() # get new ID 

65 return serial.Category.model_validate(cat) 

66 

67 

68@http 

69def put_category( 

70 session: Session, user: User, category_id: int, category_doc: serial.NewCategory 

71): 

72 """ 

73 Update the name or description for the Category at the given ID 

74 

75 This action requires the user permission 'MANAGE_ORGANISATION' 

76 """ 

77 validate.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

78 cat = fetch.category_for_user(session, user, category_id) 

79 cat.name = category_doc.name 

80 cat.description = category_doc.description 

81 

82 

83@http 

84def get_categories(session: Session, user: User) -> List[serial.Category]: 

85 """Fetch an array of all Category objects for the user's organisation""" 

86 q = ( 

87 session.query(Category) 

88 .filter(Category.organisation == user.organisation) 

89 .order_by(Category.name) 

90 ) 

91 return [c.as_dict() for c in q]