Coverage for app / services / link_service.py: 100%
93 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-21 23:51 +0300
1from datetime import UTC, datetime, timedelta
3from sqlalchemy.ext.asyncio import AsyncSession
5from app.core.config import settings
6from app.core.exceptions import (
7 AlreadyExistsError,
8 NotFoundError,
9 PermissionDeniedError
10)
11from app.models import Link
12from app.repositories.link_repository import LinkRepository
13from app.core.security import generate_short_code_n_times
15async def create_link_service(
16 db: AsyncSession,
17 long_url: str,
18 auto_prolong: bool = False,
19 alias: str | None = None,
20 expires_at: datetime | None = None,
21 user_id: int | None = None,
22) -> Link:
23 """
24 Перед созданием ссылки проверяет наличие указанной
25 короткой ссылки в базе данных.
27 - Если ссылка существует и время её жизни не истекло, то вызывается
28 исключение `AlreadyExistsError`.
29 - Если обнаружена просроченная ссылка, то она помечается на удаление
30 в текущей сессии, а занятый `alias` освобождается: пользователь
31 сможет создать новую ссылку с таким `alias`.
32 """
33 now = datetime.now(tz=UTC).replace(tzinfo=None)
35 if alias:
36 existing_link = await LinkRepository.get_by_short_code(db, alias)
37 if existing_link and existing_link.expires_at > now:
38 raise AlreadyExistsError("Alias already exists")
39 elif existing_link and existing_link.expires_at < now:
40 await LinkRepository.delete(db, existing_link)
41 await db.flush()
42 short_code = alias
43 else:
44 short_code = await generate_short_code_n_times(db)
46 # устанавливаем дату истечения действия короткой ссылки, если пользователь её не указал
47 if not expires_at:
48 expires_at = now + timedelta(
49 days=settings.link_expire_days
50 )
52 link = Link(
53 long_url=long_url,
54 user_id=user_id,
55 short_code=short_code,
56 expires_at=expires_at,
57 auto_prolong=auto_prolong
58 )
60 try:
61 db.add(link)
62 await db.commit()
63 await db.refresh(link)
64 return link
65 except Exception:
66 await db.rollback()
67 raise
69async def update_link_service(
70 db: AsyncSession,
71 current_short_code: str,
72 user_id: int,
73 alias: str | None = None
74) -> Link:
75 """
76 Обновляет короткий код для активных ссылок.
78 - Если ссылки с переданным коротким кодом не существует
79 или срок действия ссылки прошёл, то вызывается исключение `NotFoundError`.
80 - Если id пользователя, запрашивающего обновление ссылки, не совпадает с id
81 пользователя в базе данных, то вызывается исключение `PermissionDeniedError`.
82 - Если ссылка существует и время её жизни не истекло, то вызывается
83 исключение `AlreadyExistsError`.
84 - Если обнаружена просроченная ссылка, то она удаляется, а `alias` освобождается.
85 """
86 now = datetime.now(tz=UTC).replace(tzinfo=None)
87 existing_link = await LinkRepository.get_active_by_short_code(db, current_short_code, now)
88 if not existing_link:
89 raise NotFoundError(f"Short link not found")
91 if existing_link.user_id != user_id:
92 raise PermissionDeniedError()
94 if alias:
95 link = await LinkRepository.get_by_short_code(db, alias)
96 if link and link.expires_at > now:
97 raise AlreadyExistsError("Alias already exists")
98 elif link and link.expires_at < now:
99 await LinkRepository.delete(db, link)
100 await db.flush()
101 new_short_code = alias
102 else:
103 new_short_code = await generate_short_code_n_times(db)
105 try:
106 await LinkRepository.update_short_code(db, current_short_code, new_short_code)
107 await db.commit()
108 await db.refresh(existing_link)
109 return existing_link
110 except Exception:
111 await db.rollback()
112 raise
115async def redirect_link_service(db: AsyncSession, short_code: str) -> Link:
116 """
117 Перед перенаправлением пользователя на исходный URL вызывает
118 функцию из репозитория, которая обновляет дату истечения
119 действия ссылки и количество переходов по ссылке.
121 - Если ссылки с переданным коротким кодом не существует
122 или срок действия ссылки прошёл, то вызывается исключение `NotFoundError`.
123 - Если у ссылки активен флаг автопродления и до даты истечения действия
124 ссылки осталось меньше 15 дней, то время жизни ссылки увеличивается.
125 """
126 now = datetime.now(tz=UTC).replace(tzinfo=None)
127 new_exp_date = None
129 link = await LinkRepository.get_active_by_short_code(db, short_code, now)
130 if not link:
131 raise NotFoundError(f"Short link not found")
133 if link.auto_prolong and (link.expires_at - now) < timedelta(days=15):
134 new_exp_date = now + timedelta(days=settings.link_expire_days)
136 try:
137 await LinkRepository.update_on_redirect(db, short_code, now, new_exp_date)
138 await db.commit()
139 await db.refresh(link)
140 return link
141 except Exception:
142 await db.rollback()
143 raise
146async def delete_link_service(db: AsyncSession, short_code: str, user_id: int) -> None:
147 """
148 Удаляет ссылку из базы данных.
150 - Если ссылки с переданным коротким кодом не существует, то вызывается
151 исключение `NotFoundError`.
152 - Если пользователь пытается удалить чужую ссылку, то вызывается
153 исключение `PermissionDeniedError`.
154 """
155 link = await LinkRepository.get_by_short_code(db, short_code)
156 if not link:
157 raise NotFoundError(f"Short link not found")
158 if link.user_id != user_id:
159 raise PermissionDeniedError()
160 try:
161 await LinkRepository.delete(db, link)
162 await db.commit()
163 except Exception:
164 await db.rollback()
165 raise
168async def get_stats_link_service(db: AsyncSession, short_code: str) -> dict:
169 """
170 Возвращает статистику по активным ссылкам с указанным коротким кодом
171 (исходный URL, дата создания, количество переходов, дата последнего перехода).
173 - Если ссылки с переданным коротким кодом не существует
174 или срок действия ссылки прошёл, то вызывается исключение `NotFoundError`.
175 """
176 now = datetime.now(tz=UTC).replace(tzinfo=None)
177 link = await LinkRepository.get_active_by_short_code(db, short_code, now)
178 if not link:
179 raise NotFoundError(f"Short link not found")
180 return {
181 "long_url": link.long_url,
182 "created_at": link.created_at,
183 "visits_counter": link.visits_counter,
184 "last_visited_at": link.last_visited_at,
185 }
187async def get_by_long_url_link_service(
188 db: AsyncSession, long_url: str, user_id: int | None = None
189) -> list[Link]:
190 """
191 Вовзращает список коротких ссылок для указанного исходного URL.
193 - Если ссылки с переданным коротким кодом не существует
194 или срок действия ссылки прошёл, то вызывается исключение `NotFoundError`.
195 """
196 now = datetime.now(tz=UTC).replace(tzinfo=None)
197 links = await LinkRepository.get_active_by_long_url(db, long_url, now, user_id)
198 if not links:
199 raise NotFoundError("Short links for provided long url not found")
200 return links