Coverage for app/crud.py: 100%

1import uuid (empty)

2from typing import Any (empty)

3 

4from sqlmodel import Session, select (empty)

5 

6from app.core.security import get_password_hash, verify_password (empty)

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

8 

9 

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

11 db_obj = User.model_validate( 30 ctx1ajklmnopbstdABueCvDgwxfEFGHIJc

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

13 ) 

14 session.add(db_obj) 30 ctx1ajklmnopbstdABueCvDgwxfEFGHIJc

15 session.commit() 30 ctx1ajklmnopbstdABueCvDgwxfEFGHIJc

16 session.refresh(db_obj) 30 ctx1ajklmnopbstdABueCvDgwxfEFGHIJc

17 return db_obj 30 ctx1ajklmnopbstdABueCvDgwxfEFGHIJc

18 

19 

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

21 user_data = user_in.model_dump(exclude_unset=True) 4 ctx1abgc

22 extra_data = {} 4 ctx1abgc

23 if "password" in user_data: 4 ctx1abgc

24 password = user_data["password"] 3 ctx1abc

25 hashed_password = get_password_hash(password) 3 ctx1abc

26 extra_data["hashed_password"] = hashed_password 3 ctx1abc

27 db_user.sqlmodel_update(user_data, update=extra_data) 4 ctx1abgc

28 session.add(db_user) 4 ctx1abgc

29 session.commit() 4 ctx1abgc

30 session.refresh(db_user) 4 ctx1abgc

31 return db_user 4 ctx1abgc

32 

33 

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

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

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

37 return session_user 22 ctx1aqyrhKLbstMduevNwOxfiz

38 

39 

40# Dummy hash to use for timing attack prevention when user is not found 

41# This is an Argon2 hash of a random password, used to ensure constant-time comparison 

42DUMMY_HASH = "$argon2id$v=19$m=65536,t=3,p=4$MjQyZWE1MzBjYjJlZTI0Yw$YTU4NGM5ZTZmYjE2NzZlZjY0ZWY3ZGRkY2U2OWFjNjk" (empty)

43 

44 

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

46 db_user = get_user_by_email(session=session, email=email) 11 ctx1aqyrhbdefiz

47 if not db_user: 11 ctx1aqyrhbdefiz

48 # Prevent timing attacks by running password verification even when user doesn't exist 

49 # This ensures the response time is similar whether or not the email exists 

50 verify_password(password, DUMMY_HASH) 1 ctx1z

51 return None 1 ctx1z

52 verified, updated_password_hash = verify_password(password, db_user.hashed_password) 10 ctx1aqyrhbdefi

53 if not verified: 10 ctx1aqyrhbdefi

54 return None 1 ctx1y

55 if updated_password_hash: 9 ctx1aqrhbdefi

56 db_user.hashed_password = updated_password_hash 2 ctx1hi

57 session.add(db_user) 2 ctx1hi

58 session.commit() 2 ctx1hi

59 session.refresh(db_user) 2 ctx1hi

60 return db_user 9 ctx1aqrhbdefi

61 

62 

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

64 db_item = Item.model_validate(item_in, update={"owner_id": owner_id}) 7 ctx1jklmnop

65 session.add(db_item) 7 ctx1jklmnop

66 session.commit() 7 ctx1jklmnop

67 session.refresh(db_item) 7 ctx1jklmnop

68 return db_item 7 ctx1jklmnop