Coverage for src/routes/users.py: 74%

1from fastapi import APIRouter, Depends, UploadFile, File 

2 

3from fastapi_limiter.depends import RateLimiter 

4 

5from sqlalchemy.ext.asyncio import AsyncSession 

6 

7import cloudinary 

8import cloudinary.uploader 

9 

10from src.database.db import get_db 

11from src.entity.models import User 

12from src.repository import users as repositories_users 

13from src.schemas.user import UserResponseSchema 

14from src.services.auth import auth_serviсe 

15from src.conf.config import settings 

16from src.schemas.user import UserDb 

17 

18 

19router = APIRouter(prefix="/users", tags=["users"]) 

20 

21 

22@router.get( 

23 "/me", 

24 response_model=UserResponseSchema, 

25 description="No more than 2 requests per 10 seconds", 

26 dependencies=[Depends(RateLimiter(times=2, seconds=10))], 

27) 

28async def get_current_user(user: User = Depends(auth_serviсe.get_current_user)): 

29 """ 

30 Retrieves the current authenticated user's information. 

31 

32 Parameters: 

33 - user (User): The authenticated user object obtained from the auth_service. 

34 

35 Returns: 

36 - UserResponseSchema: The current authenticated user's information in the specified response model format. 

37 

38 This endpoint is protected by a rate limiter that allows no more than 2 requests per 10 seconds. 

39 """ 

40 return user 

41 

42 

43@router.patch("/avatar", response_model=UserDb) 

44async def update_avatar_user( 

45 file: UploadFile = File(), 

46 current_user: User = Depends(auth_serviсe.get_current_user), 

47 db: AsyncSession = Depends(get_db), 

48): 

49 """ 

50 Updates the avatar of the current authenticated user. 

51 

52 Parameters: 

53 - file (UploadFile): The file to be uploaded as the new avatar. 

54 - current_user (User): The authenticated user object obtained from the auth_service. 

55 - db (AsyncSession): The database session to be used for the operation. 

56 

57 Returns: 

58 - UserDb: The updated user object with the new avatar URL. 

59 

60 This method uses Cloudinary to upload the new avatar file and update the user's avatar URL in the database. 

61 """ 

62 cloudinary.config( 

63 cloud_name=settings.cloudinary_name, 

64 api_key=settings.cloudinary_api_key, 

65 api_secret=settings.cloudinary_api_secret, 

66 secure=True, 

67 ) 

68 

69 cloudinary.uploader.upload( 

70 file.file, public_id=f"ContactsApp/{current_user.username}", overwrite=True 

71 ) 

72 src_url = cloudinary.CloudinaryImage( 

73 f"ContactsApp/{current_user.username}" 

74 ).build_url(width=250, height=250, crop="fill") 

75 user = await repositories_users.update_avatar(current_user.email, src_url, db) 

76 return user