Coverage for src / links / routers.py: 96%

1from fastapi import APIRouter, Depends, HTTPException, Query, status 

2from fastapi.responses import RedirectResponse 

3from sqlalchemy.ext.asyncio import AsyncSession 

4from typing import Optional 

5 

6from src.database import get_db 

7from src.auth.manager import current_user, current_optional_user 

8from src.auth.models import User 

9from . import crud, schemas 

10from .exceptions import NotUniqueAliasError, AliasLengthError 

11 

12router = APIRouter(prefix="/links", tags=["links"]) 

13 

14 

15# 1. Создание ссылки (доступно всем) 

16@router.post("/shorten", response_model=schemas.LinkResponse) 

17async def create_short_link( 

18 link_data: schemas.LinkCreate, 

19 db: AsyncSession = Depends(get_db), 

20 user: Optional[User] = Depends(current_optional_user) 

21): 

22 try: 

23 return await crud.create_link( 

24 db=db, 

25 original_url=str(link_data.original_url), 

26 custom_alias=link_data.custom_alias, 

27 expire_at=link_data.expire_at, 

28 user_id=user.id if user else None 

29 ) 

30 except (NotUniqueAliasError, AliasLengthError) as e: 

31 raise HTTPException(status.HTTP_400_BAD_REQUEST, detail=str(e)) 

32 

33 

34# 2. Редирект (доступно всем) 

35@router.get("/{short_id}") 

36async def redirect_link( 

37 short_id: str, 

38 db: AsyncSession = Depends(get_db) 

39): 

40 link = await crud.get_link_by_short_id(db, short_id) 

41 if not link: 

42 raise HTTPException(status.HTTP_404_NOT_FOUND, detail="Link not found") 

43 await crud.increment_click_count(db, link) 

44 return RedirectResponse(url=link.original_url) 

45 

46 

47# 3. Статистика (только для авторизованных) 

48@router.get("/{short_id}/stats", response_model=schemas.LinkStatsResponse) 

49async def get_link_stats( 

50 short_id: str, 

51 db: AsyncSession = Depends(get_db), 

52 user: User = Depends(current_user) 

53): 

54 link = await crud.get_link_by_short_id(db, short_id) 

55 if not link or link.user_id != user.id: 

56 raise HTTPException(status.HTTP_404_NOT_FOUND, detail="Link not found") 

57 

58 return { 

59 "original_url": link.original_url, 

60 "short_id": link.short_id, 

61 "custom_alias": link.custom_alias, 

62 "created_at": link.created_at, 

63 "expire_at": link.expire_at, 

64 "click_count": link.click_count, 

65 } 

66 

67 

68# 4. Поиск (только для авторизованных) 

69@router.get("/search/", response_model=list[schemas.LinkResponse]) 

70async def search_links( 

71 original_url: str = Query(..., min_length=3), 

72 db: AsyncSession = Depends(get_db), 

73 user: User = Depends(current_user) 

74): 

75 return await crud.search_links(db, original_url, user.id) 

76 

77 

78# 5. Удаление (только для авторизованных) 

79@router.delete("/{short_id}") 

80async def delete_link( 

81 short_id: str, 

82 db: AsyncSession = Depends(get_db), 

83 user: User = Depends(current_user) 

84): 

85 link = await crud.get_link_by_short_id(db, short_id) 

86 if not link or link.user_id != user.id: 

87 raise HTTPException(status.HTTP_404_NOT_FOUND, detail="Link not found") 

88 

89 await crud.delete_link(db, link) 

90 return {"status": "success"} 

91 

92 

93# 6. Обновление (только для авторизованных) 

94@router.put("/{short_id}", response_model=schemas.LinkResponse) 

95async def update_link( 

96 short_id: str, 

97 update_data: schemas.LinkUpdate, 

98 db: AsyncSession = Depends(get_db), 

99 user: User = Depends(current_user) 

100): 

101 link = await crud.get_link_by_short_id(db, short_id) 

102 if not link or link.user_id != user.id: 

103 raise HTTPException(status.HTTP_404_NOT_FOUND, detail="Link not found") 

104 

105 return await crud.update_link(db, link, str(update_data.original_url))