Coverage for app / routers / link_router.py: 100%

1from typing import Annotated 

2 

3from fastapi import APIRouter, Depends, HTTPException, Path, Query, status 

4from fastapi.responses import RedirectResponse 

5from fastapi_cache.decorator import cache 

6from pydantic import HttpUrl 

7from sqlalchemy.exc import SQLAlchemyError 

8from sqlalchemy.ext.asyncio import AsyncSession 

9 

10from app.core.config import settings 

11from app.core.database import get_db 

12from app.core.exceptions import ( 

13 AlreadyExistsError, 

14 NotFoundError, 

15 PermissionDeniedError, 

16 UniqueCodeGenerationError, 

17) 

18from app.core.key_builder import redis_custom_key_builder 

19from app.core.security import get_current_user, get_current_user_optional 

20from app.models import User 

21from app.schemas.link_schemas import ( 

22 LinkSearchResponse, 

23 LinkShortenRequest, 

24 LinkShortenResponse, 

25 LinkStatsResponse, 

26 LinkUpdateRequest, 

27 LinkUpdateResponse, 

28 short_code_pattern, 

29) 

30from app.services.link_service import ( 

31 create_link_service, 

32 delete_link_service, 

33 get_by_long_url_link_service, 

34 get_stats_link_service, 

35 redirect_link_service, 

36 update_link_service, 

37) 

38 

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

40 

41@router.post("/shorten") 

42async def shorten_link( 

43 data: LinkShortenRequest, 

44 user: Annotated[User | None, Depends(get_current_user_optional)], 

45 db: Annotated[AsyncSession, Depends(get_db)], 

46) -> LinkShortenResponse: 

47 """ 

48 Вызывает сервис для создания ссылки. 

49 

50 Возвращает сообщение об успешном создании ссылки, исходный URL 

51 и короткую ссылку. 

52 """ 

53 try: 

54 link = await create_link_service( 

55 db, 

56 long_url=str(data.long_url), 

57 auto_prolong=data.auto_prolong, 

58 alias=data.custom_alias, 

59 expires_at=data.expires_at, 

60 user_id=user.id if user else None, 

61 ) 

62 except AlreadyExistsError as e: 

63 raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e)) 

64 

65 except UniqueCodeGenerationError as e: 

66 raise HTTPException( 

67 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e) 

68 ) 

69 

70 except SQLAlchemyError: 

71 raise HTTPException( 

72 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error" 

73 ) 

74 

75 return LinkShortenResponse( 

76 msg="Short link created succesfully", 

77 long_url=link.long_url, 

78 short_url=f"{str(settings.base_url).rstrip("/")}/links/{link.short_code}", 

79 ) 

80 

81 

82@router.put("/{short_code}") 

83async def update_short_link( 

84 data: LinkUpdateRequest, 

85 user: Annotated[User, Depends(get_current_user)], 

86 short_code: Annotated[ 

87 str, 

88 Path( 

89 pattern=short_code_pattern, description="URL short code or alias to update" 

90 ), 

91 ], 

92 db: Annotated[AsyncSession, Depends(get_db)], 

93) -> LinkUpdateResponse: 

94 """ 

95 Вызывает сервис для обновления короткого кода ссылки. Эндпоинт доступен 

96 только авторизованным пользователям. 

97 

98 Возвращает сообщение об успешном обновлении ссылки, исходный URL, 

99 старую и новую короткие ссылки. 

100 """ 

101 try: 

102 updated_link = await update_link_service( 

103 db, 

104 short_code, 

105 user_id=user.id, 

106 alias=data.custom_alias, 

107 ) 

108 except NotFoundError as e: 

109 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) 

110 

111 except PermissionDeniedError as e: 

112 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(e)) 

113 

114 except AlreadyExistsError as e: 

115 raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e)) 

116 

117 except UniqueCodeGenerationError as e: 

118 raise HTTPException( 

119 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e) 

120 ) 

121 

122 except SQLAlchemyError: 

123 raise HTTPException( 

124 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error" 

125 ) 

126 

127 return LinkUpdateResponse( 

128 msg="Short link updated successfully", 

129 long_url=updated_link.long_url, 

130 old_short_url=f"{str(settings.base_url).rstrip("/")}/links/{short_code}", 

131 new_short_url=f"{str(settings.base_url).rstrip("/")}/links/{updated_link.short_code}", 

132 ) 

133 

134 

135@router.get("/search") 

136@cache(expire=60, key_builder=redis_custom_key_builder) 

137async def get_links_by_long_url( 

138 user: Annotated[User | None, Depends(get_current_user_optional)], 

139 original_url: Annotated[HttpUrl, Query(description="Long URL")], 

140 db: Annotated[AsyncSession, Depends(get_db)], 

141) -> LinkSearchResponse: 

142 """ 

143 Вызывает сервис для поиска коротких ссылок, созданных для 

144 указанного исходного URL. 

145 

146 Авторизованные пользователи получают список своих коротких ссылок (по `user_id`), 

147 а неавторизованные - ссылки, которые не принадлежат ни одному из зарегистрированных 

148 пользователей. 

149 """ 

150 try: 

151 links = await get_by_long_url_link_service( 

152 db, str(original_url), user_id=user.id if user else None 

153 ) 

154 except NotFoundError as e: 

155 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) 

156 

157 return LinkSearchResponse( 

158 short_urls=[ 

159 f"{str(settings.base_url).rstrip("/")}/links/{link.short_code}" 

160 for link in links 

161 ] 

162 ) 

163 

164 

165@router.get("/{short_code}") 

166async def redirect_to_long_url( 

167 short_code: Annotated[ 

168 str, Path(pattern=short_code_pattern, description="URL short code or alias") 

169 ], 

170 db: Annotated[AsyncSession, Depends(get_db)], 

171): 

172 """ 

173 Перенаправляет пользователя на исходный URL. 

174 """ 

175 try: 

176 link = await redirect_link_service(db, short_code) 

177 long_url = link.long_url 

178 

179 except NotFoundError as e: 

180 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) 

181 

182 except SQLAlchemyError: 

183 raise HTTPException( 

184 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error" 

185 ) 

186 

187 return RedirectResponse(url=long_url) 

188 

189 

190@router.delete("/{short_code}") 

191async def delete_link( 

192 short_code: Annotated[ 

193 str, Path(pattern=short_code_pattern, description="URL short code or alias") 

194 ], 

195 user: Annotated[User, Depends(get_current_user)], 

196 db: Annotated[AsyncSession, Depends(get_db)], 

197) -> dict[str, str]: 

198 """ 

199 Вызывает сервис для удаления ссылки. Эндпоинт доступен только авторизованным пользователям. 

200 

201 Возвращает сообщение об успешном удалении ссылки и короткий код удалённой ссылки. 

202 """ 

203 try: 

204 await delete_link_service(db, short_code, user_id=user.id) 

205 

206 except NotFoundError as e: 

207 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) 

208 

209 except PermissionDeniedError as e: 

210 raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(e)) 

211 

212 except SQLAlchemyError: 

213 raise HTTPException( 

214 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error" 

215 ) 

216 

217 return { 

218 "msg": "Short link deleted successfully", 

219 "deleted_link_short_code": short_code, 

220 } 

221 

222 

223@router.get("/{short_code}/stats") 

224@cache(expire=60, key_builder=redis_custom_key_builder) 

225async def get_link_stats( 

226 short_code: Annotated[ 

227 str, Path(pattern=short_code_pattern, description="URL short code or alias") 

228 ], 

229 db: Annotated[AsyncSession, Depends(get_db)], 

230) -> LinkStatsResponse: 

231 """ 

232 Вызывает сервис для получения статистики по короткой ссылке. 

233 

234 Возвращает исходный URL, дату создания, количество переходов по ссылке 

235 и дату последнего перехода. 

236 """ 

237 try: 

238 response = await get_stats_link_service(db, short_code) 

239 except NotFoundError as e: 

240 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) 

241 

242 except SQLAlchemyError: 

243 raise HTTPException( 

244 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error" 

245 ) 

246 

247 return LinkStatsResponse(**response)