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

1from uuid import UUID 

2 

3from sqlalchemy import or_ 

4from sqlalchemy.orm import Session 

5 

6from app.models import User, UserRole 

7from app.schemas import UserCreate, UserUpdate 

8 

9 

10def get_user(db: Session, user_id: UUID) -> User | None: 

11 """Get user by ID""" 

12 return db.query(User).filter(User.userId == user_id).first() 

13 

14 

15def get_user_by_username(db: Session, username: str) -> User | None: 

16 """Get user by username""" 

17 return db.query(User).filter(User.username == username).first() 

18 

19 

20def get_user_by_email(db: Session, email: str) -> User | None: 

21 """Get user by email""" 

22 return db.query(User).filter(User.email == email).first() 

23 

24 

25def get_users( 

26 db: Session, 

27 skip: int = 0, 

28 limit: int = 100, 

29 role: UserRole | None = None, 

30 course_id: UUID | None = None 

31) -> list[User]: 

32 """Get users with optional filters""" 

33 query = db.query(User) 

34 

35 if role: 

36 query = query.filter(User.role == role) 

37 if course_id: 

38 query = query.filter(User.courseId == course_id) 

39 

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

41 

42 

43def create_user(db: Session, user: UserCreate) -> User: 

44 """Create a new user""" 

45 db_user = User( 

46 email=user.email, 

47 username=user.username, 

48 role=user.role, 

49 courseId=user.courseId 

50 ) 

51 db.add(db_user) 

52 db.commit() 

53 db.refresh(db_user) 

54 return db_user 

55 

56 

57def update_user(db: Session, user_id: UUID, user_update: UserUpdate) -> User | None: 

58 """Update user information""" 

59 db_user = get_user(db, user_id) 

60 if not db_user: 

61 return None 

62 

63 update_data = user_update.model_dump(exclude_unset=True) 

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

65 setattr(db_user, field, value) 

66 

67 db.commit() 

68 db.refresh(db_user) 

69 return db_user 

70 

71 

72 

73 

74def delete_user(db: Session, user_id: UUID) -> bool: 

75 """Delete a user""" 

76 db_user = get_user(db, user_id) 

77 if not db_user: 

78 return False 

79 

80 db.delete(db_user) 

81 db.commit() 

82 return True 

83 

84 

85def search_users(db: Session, query: str, limit: int = 10) -> list[User]: 

86 """Search users by username or email""" 

87 return db.query(User).filter( 

88 or_( 

89 User.username.ilike(f"%{query}%"), 

90 User.email.ilike(f"%{query}%") 

91 ) 

92 ).limit(limit).all()