Coverage for app/crud/teams.py: 100.00%

1from uuid import UUID 

2 

3from sqlalchemy.orm import Session 

4 

5from app.models import Team, UserToTeam 

6from app.schemas import TeamCreate, TeamUpdate 

7 

8 

9def get_team(db: Session, team_id: UUID) -> Team | None: 

10 """Get team by ID""" 

11 return db.query(Team).filter(Team.teamId == team_id).first() 

12 

13 

14def get_teams( 

15 db: Session, 

16 skip: int = 0, 

17 limit: int = 100, 

18 deployment_id: UUID | None = None 

19) -> list[Team]: 

20 """Get teams with optional deployment filter.""" 

21 query = db.query(Team) 

22 

23 if deployment_id: 

24 query = query.filter(Team.deploymentId == deployment_id) 

25 

26 return query.offset(skip).limit(limit).all() 

27 

28 

29def _add_team_members(db: Session, team: Team, user_ids: list[UUID]) -> None: 

30 """Stage ``UserToTeam`` membership rows for ``team``. 

31 

32 Adds one association row per user id to the session without 

33 committing or flushing — the caller controls transaction 

34 boundaries. ``team.teamId`` must already be populated (via a prior 

35 commit or flush) so the foreign key can be set. 

36 """ 

37 for user_id in user_ids: 

38 user_to_team = UserToTeam( 

39 userId=user_id, 

40 teamId=team.teamId 

41 ) 

42 db.add(user_to_team) 

43 

44 

45def create_team(db: Session, team: TeamCreate) -> Team: 

46 """Create a new team. 

47 

48 ``Team`` has a NOT NULL ``deploymentId`` FK, so the request payload 

49 must carry the deployment to attach to. 

50 """ 

51 db_team = Team( 

52 name=team.name, 

53 deploymentId=team.deploymentId 

54 ) 

55 db.add(db_team) 

56 # Commit the team row first so ``teamId`` is populated before staging 

57 # memberships. 

58 db.commit() 

59 db.refresh(db_team) 

60 

61 _add_team_members(db, db_team, team.userIds) 

62 

63 db.commit() 

64 db.refresh(db_team) 

65 return db_team 

66 

67 

68def update_team(db: Session, team_id: UUID, team_update: TeamUpdate) -> Team | None: 

69 """Update team information""" 

70 db_team = get_team(db, team_id) 

71 if not db_team: 

72 return None 

73 

74 update_data = team_update.model_dump(exclude_unset=True) 

75 for field, value in update_data.items(): 

76 setattr(db_team, field, value) 

77 

78 db.commit() 

79 db.refresh(db_team) 

80 return db_team 

81 

82 

83def delete_team(db: Session, team_id: UUID) -> bool: 

84 """Delete a team""" 

85 db_team = get_team(db, team_id) 

86 if not db_team: 

87 return False 

88 

89 db.delete(db_team) 

90 db.commit() 

91 return True 

92 

93 

94def add_user_to_team(db: Session, team_id: UUID, user_id: UUID) -> bool: 

95 """Add a user to a team""" 

96 # Check if already exists 

97 existing = db.query(UserToTeam).filter( 

98 UserToTeam.teamId == team_id, 

99 UserToTeam.userId == user_id 

100 ).first() 

101 

102 if existing: 

103 return False 

104 

105 user_to_team = UserToTeam( 

106 userId=user_id, 

107 teamId=team_id 

108 ) 

109 db.add(user_to_team) 

110 db.commit() 

111 return True 

112 

113 

114def remove_user_from_team(db: Session, team_id: UUID, user_id: UUID) -> bool: 

115 """Remove a user from a team""" 

116 user_to_team = db.query(UserToTeam).filter( 

117 UserToTeam.teamId == team_id, 

118 UserToTeam.userId == user_id 

119 ).first() 

120 

121 if not user_to_team: 

122 return False 

123 

124 db.delete(user_to_team) 

125 db.commit() 

126 return True 

127 

128 

129def create_teams_for_deployment( 

130 db: Session, 

131 deployment_id: UUID, 

132 teams_data: list[dict] 

133) -> list[Team]: 

134 """ 

135 Create multiple teams for a deployment 

136 teams_data format: [{"name": "team1", "userIds": [uuid1, uuid2]}, ...] 

137 """ 

138 created_teams = [] 

139 

140 for team_data in teams_data: 

141 # Create team 

142 db_team = Team( 

143 name=team_data["name"], 

144 deploymentId=deployment_id 

145 ) 

146 db.add(db_team) 

147 db.flush() # Get team ID 

148 

149 _add_team_members(db, db_team, team_data.get("userIds", [])) 

150 

151 created_teams.append(db_team) 

152 

153 return created_teams