add file

Author: chaarse

auth.py

@@ -1,30 +1,23 @@
-import os
 import logging
 import hashlib
-from datetime import datetime
 from typing import Optional, Dict
-
-from fastapi import APIRouter, Depends, HTTPException, status, Form
+from fastapi import APIRouter, Depends, HTTPException, Form
 from fastapi.security import OAuth2PasswordBearer
 from sqlalchemy import select
 from passlib.context import CryptContext
-
 from database import new_session, UserOrm, LinkOrm
 from schemas import UserRegister, UserResponse
 
 logger = logging.getLogger(__name__)
 
-# Контекст для хэширования паролей
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
-# Схема OAuth2 (без токенов, просто для совместимости)
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token", auto_error=False)
 
-# Хранилище авторизованных пользователей (в памяти)
 active_users: Dict[str, UserResponse] = {}
 
-# Создаем роутер для аутентификации
-auth_router = APIRouter(prefix="/auth", tags=["Аутентификация"])
+auth_router = APIRouter(prefix="/auth",
+                        tags=["Аутентификация"])
 
 
 def generate_user_secret_key(username: str) -> str:
@@ -36,19 +29,16 @@ class AuthService:
     @classmethod
     async def register_user(cls, user_data: UserRegister) -> UserResponse:
         """
-        Регистрирует нового пользователя.
+        Регистрация пользователя.
         """
         async with new_session() as session:
             try:
-                # Проверяем, существует ли пользователь
                 existing_user = await session.execute(select(UserOrm).where(UserOrm.username == user_data.username))
                 if existing_user.scalar():
                     raise HTTPException(status_code=400, detail="Username already exists")
 
-                # Хэшируем пароль
                 hashed_password = pwd_context.hash(user_data.password)
 
-                # Создаем нового пользователя
                 user = UserOrm(username=user_data.username, password_hash=hashed_password)
                 session.add(user)
                 await session.flush()
@@ -61,10 +51,11 @@ async def register_user(cls, user_data: UserRegister) -> UserResponse:
                 await session.rollback()
                 raise HTTPException(status_code=500, detail="Internal Server Error")
 
+
     @classmethod
     async def authenticate_user(cls, username: str, password: str) -> UserOrm:
         """
-        Аутентифицирует пользователя.
+        Аутентификация пользователя.
         """
         async with new_session() as session:
             user = await session.execute(select(UserOrm).where(UserOrm.username == username))
@@ -73,16 +64,14 @@ async def authenticate_user(cls, username: str, password: str) -> UserOrm:
                 raise HTTPException(status_code=401, detail="Invalid username or password")
             return user
 
+
     @classmethod
     async def get_current_user(cls, token: Optional[str] = Depends(oauth2_scheme)) -> Optional[UserResponse]:
-        """
-        Получает текущего пользователя из хранилища активных пользователей.
-        """
+
         if token is None:
             logger.debug("Токен отсутствует")
             return None
 
-        # Ищем пользователя в хранилище активных пользователей
         user = active_users.get(token)
         if user is None:
             logger.debug(f"Пользователь с токеном {token} не найден")
@@ -91,19 +80,18 @@ async def get_current_user(cls, token: Optional[str] = Depends(oauth2_scheme)) -
         logger.debug(f"Пользователь {user.username} успешно авторизован")
         return user
 
+
     @classmethod
     async def update_user_id_for_links(cls, username: str, user_id: int):
         """
-        Обновляет user_id для всех ссылок, созданных до авторизации.
+        Обновление user_id для всех ссылок, созданных до авторизации.
         """
         async with new_session() as session:
             try:
-                # Находим все ссылки, созданные до авторизации (user_id = NULL)
                 query = select(LinkOrm).where(LinkOrm.user_id.is_(None))
                 result = await session.execute(query)
                 links = result.scalars().all()
 
-                # Обновляем user_id для найденных ссылок
                 for link in links:
                     link.user_id = user_id
 
@@ -115,41 +103,32 @@ async def update_user_id_for_links(cls, username: str, user_id: int):
                 raise HTTPException(status_code=500, detail="Internal Server Error")
 
 
-# Эндпоинты для аутентификации
 @auth_router.post("/register")
 async def register(
-    username: str = Form(...),  # Ввод через форму
-    password: str = Form(...),  # Ввод через форму
+    username: str = Form(...),
+    password: str = Form(...),
 ):
-    """
-    Регистрирует нового пользователя (через форму).
-    """
+
     user_data = UserRegister(username=username, password=password)
     return await AuthService.register_user(user_data)
 
+
 @auth_router.post("/token")
 async def login_for_access_token(
-    username: str = Form(...),  # Ввод через форму
-    password: str = Form(...),  # Ввод через форму
+    username: str = Form(...),
+    password: str = Form(...),
 ):
-    """
-    Аутентифицирует пользователя и возвращает токен (просто строку).
-    Также обновляет user_id для всех ссылок, созданных до авторизации.
-    """
+
     user = await AuthService.authenticate_user(username, password)
     user_response = UserResponse(id=user.id, username=user.username)
 
-    # Генерируем "токен" (просто хэш имени пользователя)
     token = generate_user_secret_key(username)
 
-    # Сохраняем пользователя в хранилище активных пользователей
     active_users[token] = user_response
 
-    # Обновляем user_id для всех ссылок, созданных до авторизации
     await AuthService.update_user_id_for_links(username, user.id)
 
     return {"access_token": token, "token_type": "bearer"}
 
 
-# Экспортируем функцию для использования в других модулях
-get_current_user = AuthService.get_current_user
+get_current_user = AuthService.get_current_user

cache.py

@@ -1,5 +1,4 @@
 import redis
-from fastapi import HTTPException
 
 redis_client = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
 

database.py

@@ -3,19 +3,23 @@
 from datetime import datetime, timedelta
 from typing import Optional
 
+
 engine = create_async_engine("sqlite+aiosqlite:///links.db")
 new_session = async_sessionmaker(engine, expire_on_commit=False)
 
+
 class Model(DeclarativeBase):
     pass
 
+
 class UserOrm(Model):
     __tablename__ = "users"
 
     id: Mapped[int] = mapped_column(primary_key=True)
     username: Mapped[str] = mapped_column(unique=True)
     password_hash: Mapped[str]
 
+
 class LinkOrm(Model):
     __tablename__ = "links"
 
@@ -28,10 +32,12 @@ class LinkOrm(Model):
     click_count: Mapped[int] = mapped_column(default=0)
     last_used_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
 
+
 async def create_tables():
     async with engine.begin() as conn:
         await conn.run_sync(Model.metadata.create_all)
 
+
 async def delete_tables():
     async with engine.begin() as conn:
-        await conn.run_sync(Model.metadata.drop_all)
+        await conn.run_sync(Model.metadata.drop_all)

main.py

@@ -1,6 +1,5 @@
 from fastapi import FastAPI
 from fastapi.openapi.utils import get_openapi
-from contextlib import asynccontextmanager
 from database import create_tables, delete_tables
 from router import router as links_router
 from auth import auth_router
@@ -12,14 +11,14 @@
 logging.basicConfig(level=logging.DEBUG)
 logger = logging.getLogger(__name__)
 
+
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     await delete_tables()
     logger.info("База очищена")
     await create_tables()
     logger.info("База готова к работе")
 
-    # Запускаем фоновую задачу
     asyncio.create_task(delete_expired_links())
 
     yield
@@ -29,6 +28,7 @@ async def lifespan(app: FastAPI):
 app.include_router(links_router)
 app.include_router(auth_router)
 
+
 def custom_openapi():
     if app.openapi_schema:
         return app.openapi_schema

repository.py

@@ -12,33 +12,30 @@
 
 logger = logging.getLogger(__name__)
 
+
 def normalize_url(url: str) -> str:
-    """
-    Нормализует URL: декодирует и приводит к нижнему регистру.
-    """
     return unquote(url).lower().strip()
 
+
 class LinkRepository:
     @staticmethod
     def generate_short_code(length: int = 8) -> str:
         """
-        Генерирует уникальный короткий код.
+        Генерация короткого кода для URL.
         """
-        chars = string.ascii_letters + string.digits  # Буквы и цифры
+        chars = string.ascii_letters + string.digits
         return ''.join(secrets.choice(chars) for _ in range(length))
 
+
     @classmethod
     async def add_one(cls, data: SLinkAdd, user_id: Optional[int] = None) -> SLinkResponse:
         """
-        Добавляет новую ссылку в базу данных.
-        Поддерживает использование custom_alias и expires_at.
+        Добавление новой ссылки в БД.
         """
         async with new_session() as session:
             try:
-                # Нормализуем оригинальный URL
                 normalized_url = normalize_url(str(data.original_url))
 
-                # Если передан custom_alias, проверяем его уникальность
                 if data.custom_alias:
                     existing_link = await cls.find_by_short_code(data.custom_alias)
                     if existing_link:
@@ -48,22 +45,19 @@ async def add_one(cls, data: SLinkAdd, user_id: Optional[int] = None) -> SLinkRe
                         )
                     short_code = data.custom_alias
                 else:
-                    # Генерируем уникальный short_code, если custom_alias не передан
                     while True:
                         short_code = cls.generate_short_code()
                         existing_link = await cls.find_by_short_code(short_code)
                         if not existing_link:
                             break
 
-                # Устанавливаем expires_at (по умолчанию 30 дней с текущего момента)
                 expires_at = data.expires_at if data.expires_at else datetime.utcnow() + timedelta(days=30)
 
-                # Создаем новую ссылку
                 link = LinkOrm(
-                    original_url=normalized_url,  # Сохраняем нормализованный URL
+                    original_url=normalized_url,
                     short_code=short_code,
-                    user_id=user_id,  # user_id может быть None (анонимный пользователь)
-                    expires_at=expires_at,  # Устанавливаем срок действия
+                    user_id=user_id,
+                    expires_at=expires_at,
                 )
                 session.add(link)
                 await session.flush()
@@ -78,42 +72,42 @@ async def add_one(cls, data: SLinkAdd, user_id: Optional[int] = None) -> SLinkRe
                     expires_at=link.expires_at,
                     user_id=link.user_id,
                     click_count=link.click_count,
-                    short_url=None,  # Поле short_url не используется в этой ручке
+                    short_url=None,
                 )
             except HTTPException as e:
-                raise e  # Пробрасываем HTTPException
+                raise e
             except Exception as e:
                 logger.error(f"Error adding link: {e}")
                 await session.rollback()
                 raise HTTPException(status_code=500, detail="Internal Server Error")
 
+
     @classmethod
     async def find_by_short_code(cls, short_code: str) -> LinkOrm:
         """
-        Ищет ссылку по короткому коду.
+        Поиск по короткому коду.
         """
         async with new_session() as session:
             query = select(LinkOrm).where(LinkOrm.short_code == short_code)
             result = await session.execute(query)
             return result.scalars().first()
 
+
     @classmethod
     async def find_by_original_url(cls, original_url: str) -> Optional[SLinkResponse]:
         """
-        Ищет ссылку по оригинальному URL и возвращает её в формате SLinkResponse.
+        Поиск по оригинальному URL.
         """
         async with new_session() as session:
-            # Нормализуем URL
             normalized_url = normalize_url(original_url)
-            logger.debug(f"Normalized URL: {normalized_url}")  # Логируем нормализованный URL
+            logger.debug(f"Normalized URL: {normalized_url}")
 
-            # Ищем ссылку по нормализованному URL
             query = select(LinkOrm).where(LinkOrm.original_url == normalized_url)
             result = await session.execute(query)
             link = result.scalars().first()
 
             if link:
-                logger.debug(f"Found link: {link.original_url}")  # Логируем URL из базы данных
+                logger.debug(f"Found link: {link.original_url}")
             else:
                 logger.debug("Link not found")
 
@@ -128,13 +122,14 @@ async def find_by_original_url(cls, original_url: str) -> Optional[SLinkResponse
                 expires_at=link.expires_at,
                 user_id=link.user_id,
                 click_count=link.click_count,
-                short_url=f"http://127.0.0.1:8000/links/{link.short_code}",  # Формируем короткий URL
+                short_url=f"http://127.0.0.1:8000/links/{link.short_code}",
             )
 
+
     @classmethod
     async def delete_by_short_code(cls, short_code: str, user_id: int):
         """
-        Удаляет ссылку по короткому коду.
+        Удаление ссылки по короткому коду.
         """
         async with new_session() as session:
             query = delete(LinkOrm).where(
@@ -143,24 +138,22 @@ async def delete_by_short_code(cls, short_code: str, user_id: int):
             await session.execute(query)
             await session.commit()
 
+
     @classmethod
     async def update_original_url(cls, short_code: str, new_url: str, user_id: int) -> LinkOrm:
         """
-        Обновляет оригинальный URL для ссылки и возвращает обновленную запись.
+        Обновление оригинального URL.
         """
         async with new_session() as session:
             try:
-                # Нормализуем новый URL
                 normalized_url = normalize_url(new_url)
 
-                # Обновляем оригинальный URL
                 query = update(LinkOrm).where(
                     (LinkOrm.short_code == short_code) & (LinkOrm.user_id == user_id)
-                ).values(original_url=normalized_url)  # Сохраняем нормализованный URL
+                ).values(original_url=normalized_url)
                 await session.execute(query)
                 await session.commit()
 
-                # Получаем обновленную запись
                 updated_link = await cls.find_by_short_code(short_code)
                 if not updated_link:
                     logger.error(f"Failed to fetch updated link: short_code={short_code}")
@@ -172,24 +165,25 @@ async def update_original_url(cls, short_code: str, new_url: str, user_id: int)
                 await session.rollback()
                 return None
 
+
     @classmethod
     async def increment_click_count(cls, link_id: int):
         """
-        Увеличивает счетчик переходов по ссылке.
+        Счетчик переходов по ссылке.
         """
         async with new_session() as session:
             query = update(LinkOrm).where(LinkOrm.id == link_id).values(click_count=LinkOrm.click_count + 1)
             await session.execute(query)
             await session.commit()
 
+
 async def delete_expired_links():
     """
     Фоновая задача для удаления истекших ссылок.
     """
     while True:
         async with new_session() as session:
             try:
-                # Удаляем ссылки, у которых expires_at меньше текущего времени
                 query = delete(LinkOrm).where(LinkOrm.expires_at < datetime.utcnow())
                 await session.execute(query)
                 await session.commit()
@@ -198,5 +192,4 @@ async def delete_expired_links():
                 logger.error(f"Error deleting expired links: {e}")
                 await session.rollback()
 
-        # Проверяем каждые 30 минут
         await asyncio.sleep(1800)

router.py

@@ -16,35 +16,32 @@
 
 @router.get("/search", response_model=SLinkResponse)
 async def search_link_by_original_url(
-    original_url: str,  # Параметр запроса
-    request: Request,   # Для формирования короткого URL
+    original_url: str,
+    request: Request,
 ):
     """
     Поиск ссылки по оригинальному URL.
-    Возвращает ссылку с short_code.
     """
     link = await LinkRepository.find_by_original_url(original_url)
     if not link:
         raise HTTPException(status_code=404, detail="Ссылка не найдена")
 
-    # Обновляем short_url с учетом текущего хоста
     link.short_url = f"{request.base_url}links/{link.short_code}"
     return link
 
+
 @router.post("/shorten", response_model=SLinkResponse)
 async def shorten_link(
     request: Request,
     original_url: str = Form(...),
     custom_alias: Optional[str] = Form(None),
     expires_at: Optional[datetime] = Form(None),
-    user: Optional[UserResponse] = Depends(get_current_user),  # Опциональная зависимость
+    user: Optional[UserResponse] = Depends(get_current_user),
 ) -> SLinkResponse:
     """
-    Создает короткую ссылку для оригинального URL.
-    Поддерживает использование custom_alias и expires_at.
+    Создание короткой ссылки для оригинального URL.
     """
     try:
-        # Если пользователь авторизован, используем его user_id, иначе None
         user_id = user.id if user else None
         link_data = SLinkAdd(original_url=original_url, custom_alias=custom_alias, expires_at=expires_at)
         link = await LinkRepository.add_one(link_data, user_id=user_id)
@@ -59,15 +56,16 @@ async def shorten_link(
             short_url=f"{request.base_url}links/{link.short_code}",
         )
     except HTTPException as e:
-        raise e  # Пробрасываем HTTPException
+        raise e
     except Exception as e:
         logger.error(f"Error creating link: {e}")
         raise HTTPException(status_code=500, detail="Internal Server Error")
 
+
 @router.get("/{short_code}")
 async def redirect_link(short_code: str):
     """
-    Перенаправляет на оригинальный URL по короткой ссылке.
+    Перенаправление на оригинальный URL по короткой ссылке.
     """
     link = await LinkRepository.find_by_short_code(short_code)
     if not link:
@@ -76,14 +74,14 @@ async def redirect_link(short_code: str):
     await LinkRepository.increment_click_count(link.id)
     return RedirectResponse(url=link.original_url)
 
+
 @router.delete("/{short_code}")
 async def delete_link(
     short_code: str,
     user: Optional[UserResponse] = Depends(get_current_user),
 ):
     """
-    Удаляет короткую ссылку.
-    Доступно только авторизованным пользователям, которые создали ссылку.
+    Удаление короткой ссылки.
     """
     if user is None:
         raise HTTPException(status_code=403, detail="Необходима авторизация для удаления ссылки")
@@ -98,15 +96,15 @@ async def delete_link(
     await LinkRepository.delete_by_short_code(short_code, user.id)
     return {"ok": True}
 
+
 @router.put("/{short_code}", response_model=SLinkResponse)
 async def update_link(
-    short_code: str,  # short_code берется из URL
-    new_url: str = Form(...),  # new_url берется из формы (x-www-form-urlencoded)
+    short_code: str,
+    new_url: str = Form(...),
     user: Optional[UserResponse] = Depends(get_current_user),
 ) -> SLinkResponse:
     """
-    Обновляет оригинальный URL для короткой ссылки.
-    Доступно только авторизованным пользователям, которые создали ссылку.
+    Обновление оригинального URL для короткой ссылки.
     """
     if user is None:
         raise HTTPException(status_code=403, detail="Необходима авторизация для изменения ссылки")
@@ -130,14 +128,14 @@ async def update_link(
         expires_at=updated_link.expires_at,
         user_id=updated_link.user_id,
         click_count=updated_link.click_count,
-        short_url=None,  # Поле short_url не используется в этой ручке
+        short_url=None,
     )
 
+
 @router.get("/{short_code}/stats", response_model=SLinkStatsResponse)
 async def link_stats(short_code: str) -> SLinkStatsResponse:
     """
-    Возвращает статистику по короткой ссылке.
-    Доступно всем.
+    Статистика по короткой ссылке.
     """
     link = await LinkRepository.find_by_short_code(short_code)
     if not link:

schemas.py

@@ -2,23 +2,27 @@
 from datetime import datetime
 from typing import Optional
 
+
 class UserRegister(BaseModel):
     username: str = Field(..., min_length=3, max_length=50)
     password: str = Field(..., min_length=6)
 
+
 class UserLogin(BaseModel):
     username: str
     password: str
 
+
 class UserResponse(BaseModel):
     id: int
     username: str
 
     class Config:
         from_attributes = True
 
+
 class SLinkAdd(BaseModel):
-    original_url: str  # Используем str вместо HttpUrl
+    original_url: str
     custom_alias: Optional[str] = Field(
         None,
         min_length=4,
@@ -31,25 +35,28 @@ class SLinkAdd(BaseModel):
         description="Дата и время истечения срока действия ссылки (формат: YYYY-MM-DDTHH:MM)."
     )
 
+
 class SLinkResponse(BaseModel):
     id: int
     original_url: HttpUrl
     short_code: str
     created_at: datetime
     expires_at: datetime
-    user_id: Optional[int]  # ID пользователя, создавшего ссылку
+    user_id: Optional[int]
     click_count: int
-    short_url: Optional[str]  # Короткий URL
+    short_url: Optional[str]
 
     class Config:
         from_attributes = True
 
+
 class SLinkStatsResponse(BaseModel):
-    original_url: HttpUrl  # Оригинальный URL
-    created_at: datetime   # Дата создания
-    click_count: int       # Количество переходов
+    original_url: HttpUrl
+    created_at: datetime
+    click_count: int
     last_used_at: datetime
 
+
 class Token(BaseModel):
     access_token: str
     token_type: str