Coverage for app / repositories / link_repository.py: 100%

1from datetime import datetime 

2 

3from sqlalchemy import select, update 

4from sqlalchemy.ext.asyncio import AsyncSession 

5 

6from app.models import Link 

7 

8 

9class LinkRepository: 

10 @staticmethod 

11 async def get_by_short_code(db: AsyncSession, short_code: str) -> Link | None: 

12 """ 

13 Ищет ссылку по короткому коду. 

14 """ 

15 query = select(Link).where(Link.short_code == short_code) 

16 result = await db.execute(query) 

17 return result.scalar_one_or_none() 

18 

19 @staticmethod 

20 async def get_active_by_short_code( 

21 db: AsyncSession, short_code: str, now: datetime 

22 ) -> Link | None: 

23 """ 

24 По короткому коду ищет ссылку, время жизни которой ещё не истекло. 

25 """ 

26 query = select(Link).where( 

27 (Link.short_code == short_code) & (Link.expires_at > now) 

28 ) 

29 result = await db.execute(query) 

30 return result.scalar_one_or_none() 

31 

32 @staticmethod 

33 async def get_active_by_long_url( 

34 db: AsyncSession, long_url: str, now: datetime, user_id: int | None = None 

35 ) -> list[Link] | None: 

36 """ 

37 По оригинальному URL ищет активные ссылки, принадлежащие пользователю 

38 с указанным идентификатором. 

39 """ 

40 query = select(Link).where( 

41 ( 

42 (Link.long_url == long_url) 

43 & (Link.expires_at > now) 

44 & (Link.user_id == user_id) 

45 ) 

46 ) 

47 result = await db.execute(query) 

48 return result.scalars().all() 

49 

50 @staticmethod 

51 async def update_short_code( 

52 db: AsyncSession, current_short_code: str, new_short_code: str 

53 ) -> None: 

54 """ 

55 Обновляет уникальный короткий код ссылки. 

56 """ 

57 query = ( 

58 update(Link) 

59 .where(Link.short_code == current_short_code) 

60 .values(short_code=new_short_code) 

61 ) 

62 await db.execute(query) 

63 

64 @staticmethod 

65 async def update_on_redirect( 

66 db: AsyncSession, 

67 short_code: str, 

68 last_visited_at: datetime, 

69 expires_at: datetime | None = None, 

70 ) -> None: 

71 """ 

72 Увеличивает количество переходов по ссылке на единицу, обновляет 

73 дату последнего перехода по ссылке и опционально устанавливает 

74 новую дату истечения действия ссылки. 

75 """ 

76 values = { 

77 "visits_counter": Link.visits_counter + 1, 

78 "last_visited_at": last_visited_at, 

79 } 

80 if expires_at: 

81 values["expires_at"] = expires_at 

82 

83 query = update(Link).where(Link.short_code == short_code).values(**values) 

84 await db.execute(query) 

85 

86 @staticmethod 

87 async def delete(db: AsyncSession, link: Link) -> None: 

88 """ 

89 Отмечает объект ссылки `Link` как удалённый. 

90 """ 

91 await db.delete(link)