Coverage for postrfp/buyer/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 postrfp.shared import fetch 

10from postrfp.shared.decorators import http 

11from postrfp.shared import serial 

12from postrfp.authorisation import perms 

13from postrfp.model import Category, User 

14from postrfp.model.exc import DuplicateDataProvided 

15from postrfp.buyer.api import authorise 

16 

17 

18@http 

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

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

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

22 

23 

24@http 

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

26 """ 

27 Delete the category with the given ID. 

28 

29 Note that projects belong to this category are __not__ deleted 

30 This action requires the user permission 'MANAGE_ORGANISATION' 

31 """ 

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

33 authorise.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

34 session.delete(cat) 

35 

36 

37@http 

38def post_category( 

39 session: Session, user: User, category_doc: serial.NewCategory 

40) -> serial.Category: 

41 """ 

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

43 

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

45 

46 @permission MANAGE_CATEGORIES 

47 """ 

48 authorise.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

49 cat_name = category_doc.name 

50 existing = ( 

51 session.query(Category) 

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

53 .filter(Category.name == cat_name) 

54 .count() 

55 ) 

56 

57 if existing: 

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

59 

60 cat = Category( 

61 name=cat_name, 

62 description=category_doc.description, 

63 organisation=user.organisation, 

64 ) 

65 

66 session.add(cat) 

67 session.flush() # get new ID 

68 return serial.Category.model_validate(cat) 

69 

70 

71@http 

72def put_category( 

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

74): 

75 """ 

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

77 

78 This action requires the user permission 'MANAGE_ORGANISATION' 

79 """ 

80 authorise.check(user, perms.MANAGE_ORGANISATION, target_org=user.organisation) 

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

82 cat.name = category_doc.name 

83 cat.description = category_doc.description 

84 

85 

86@http 

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

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

89 q = ( 

90 session.query(Category) 

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

92 .order_by(Category.name) 

93 ) 

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