Coverage for app/crud.py: 100%

1import uuid 

2from typing import Any 

3 

4from sqlmodel import Session, select 

5 

6from app.core.security import get_password_hash, verify_password 

7from app.models import Item, ItemCreate, User, UserCreate, UserUpdate 

8 

9 

10def create_user(*, session: Session, user_create: UserCreate) -> User: 

11 db_obj = User.model_validate( 

12 user_create, update={"hashed_password": get_password_hash(user_create.password)} 

13 ) 

14 session.add(db_obj) 

15 session.commit() 

16 session.refresh(db_obj) 

17 return db_obj 

18 

19 

20def update_user(*, session: Session, db_user: User, user_in: UserUpdate) -> Any: 

21 user_data = user_in.model_dump(exclude_unset=True) 

22 extra_data = {} 

23 if "password" in user_data: 

24 password = user_data["password"] 

25 hashed_password = get_password_hash(password) 

26 extra_data["hashed_password"] = hashed_password 

27 db_user.sqlmodel_update(user_data, update=extra_data) 

28 session.add(db_user) 

29 session.commit() 

30 session.refresh(db_user) 

31 return db_user 

32 

33 

34def get_user_by_email(*, session: Session, email: str) -> User | None: 

35 statement = select(User).where(User.email == email) 

36 session_user = session.exec(statement).first() 

37 return session_user 

38 

39 

40def authenticate(*, session: Session, email: str, password: str) -> User | None: 

41 db_user = get_user_by_email(session=session, email=email) 

42 if not db_user: 

43 return None 

44 if not verify_password(password, db_user.hashed_password): 

45 return None 

46 return db_user 

47 

48 

49def create_item(*, session: Session, item_in: ItemCreate, owner_id: uuid.UUID) -> Item: 

50 db_item = Item.model_validate(item_in, update={"owner_id": owner_id}) 

51 session.add(db_item) 

52 session.commit() 

53 session.refresh(db_item) 

54 return db_item