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

1""" 

2Use Categories to organize Projects 

3""" 

4from typing import List 

5 

6from sqlalchemy.orm import Session 

7 

8from rfpy.suxint import http 

9from rfpy.web import serial 

10from rfpy.auth import perms 

11from rfpy.model import Category, User 

12from rfpy.model.exc import DuplicateDataProvided 

13from rfpy.api import fetch, validate 

14 

15 

16@http 

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

18 '''Fetch the Category object for the given ID''' 

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

20 

21 

22@http 

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

24 ''' 

25 Delete the category with the given ID. 

26 

27 Note that projects belong to this category are __not__ deleted 

28 This action requires the user permission 'MANAGE_ORGANISATION' 

29 ''' 

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

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

32 session.delete(cat) 

33 

34 

35@http 

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

37 ''' 

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

39 

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

41 

42 @permission MANAGE_CATEGORIES 

43 ''' 

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

45 cat_name = category_doc['name'] 

46 existing = session.query(Category)\ 

47 .filter(Category.organisation == user.organisation)\ 

48 .filter(Category.name == cat_name)\ 

49 .count() 

50 

51 if existing: 

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

53 

54 cat = Category( 

55 name=cat_name, 

56 description=category_doc.get("description", None), 

57 organisation=user.organisation, 

58 ) 

59 

60 session.add(cat) 

61 session.flush() # get new ID 

62 return serial.Category.from_orm(cat) 

63 

64 

65@http 

66def put_category(session: Session, user: User, category_id: int, category_doc: dict): 

67 ''' 

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

69 

70 This action requires the user permission 'MANAGE_ORGANISATION' 

71 ''' 

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

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

74 cat.name = category_doc["name"] 

75 cat.description = category_doc["description"] 

76 

77 

78@http 

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

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

81 q = (session.query(Category) 

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

83 .order_by(Category.name)) 

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

85