Coverage for app\routers\identity.py: 32%

1from fastapi import APIRouter, Depends, HTTPException 

2from sqlalchemy.orm import Session 

3from typing import List 

4from .. import db as models 

5from ..schemas import identity 

6from ..db import SessionLocal 

7from ..deps import get_db 

8 

9router = APIRouter() 

10 

11@router.post("/global-users", response_model=identity.GlobalUser) 

12def create_global_user(user_in: identity.GlobalUserCreate, db: Session = Depends(get_db)): 

13 # Check if email exists 

14 if user_in.email: 

15 existing = db.query(models.GlobalUser).filter(models.GlobalUser.email == user_in.email).first() 

16 if existing: 

17 raise HTTPException(status_code=400, detail="Email already registered") 

18 

19 user = models.GlobalUser(**user_in.model_dump()) 

20 db.add(user) 

21 db.commit() 

22 db.refresh(user) 

23 return user 

24 

25@router.get("/global-users", response_model=List[identity.GlobalUser]) 

26def read_global_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): 

27 users = db.query(models.GlobalUser).offset(skip).limit(limit).all() 

28 return users 

29 

30@router.get("/global-users/{user_id}", response_model=identity.GlobalUser) 

31def read_global_user(user_id: str, db: Session = Depends(get_db)): 

32 user = db.query(models.GlobalUser).filter(models.GlobalUser.id == user_id).first() 

33 if not user: 

34 raise HTTPException(status_code=404, detail="User not found") 

35 return user 

36 

37@router.post("/global-users/{user_id}/link", response_model=identity.GlobalUser) 

38def link_external_identity(user_id: str, system: str, external_id: str, db: Session = Depends(get_db)): 

39 user = db.query(models.GlobalUser).filter(models.GlobalUser.id == user_id).first() 

40 if not user: 

41 raise HTTPException(status_code=404, detail="User not found") 

42 

43 current_ids = dict(user.external_ids or {}) 

44 current_ids[system] = external_id 

45 user.external_ids = current_ids 

46 

47 db.commit() 

48 db.refresh(user) 

49 return user 

50 

51@router.put("/global-users/{user_id}", response_model=identity.GlobalUser) 

52def update_global_user(user_id: str, user_in: identity.GlobalUserCreate, db: Session = Depends(get_db)): 

53 user = db.query(models.GlobalUser).filter(models.GlobalUser.id == user_id).first() 

54 if not user: 

55 raise HTTPException(status_code=404, detail="User not found") 

56 

57 # Check email uniqueness if changed 

58 if user_in.email and user_in.email != user.email: 

59 existing = db.query(models.GlobalUser).filter(models.GlobalUser.email == user_in.email).first() 

60 if existing: 

61 raise HTTPException(status_code=400, detail="Email already registered") 

62 

63 update_data = user_in.model_dump(exclude_unset=True) 

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

65 setattr(user, field, value) 

66 

67 db.commit() 

68 db.refresh(user) 

69 return user 

70 

71@router.delete("/global-users/{user_id}") 

72def delete_global_user(user_id: str, db: Session = Depends(get_db)): 

73 user = db.query(models.GlobalUser).filter(models.GlobalUser.id == user_id).first() 

74 if not user: 

75 raise HTTPException(status_code=404, detail="User not found") 

76 

77 db.delete(user) 

78 db.commit() 

79 return {"ok": True}