68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
from typing import List, Optional
|
|
|
|
from fastapi import Depends, HTTPException, Path, Query
|
|
from starlette import status
|
|
|
|
from app.api.dependencies.authentication import get_current_user_authorizer
|
|
from app.api.dependencies.database import get_repository
|
|
from app.db.errors import EntityDoesNotExist
|
|
from app.db.repositories.articles import ArticlesRepository
|
|
from app.models.domain.articles import Article
|
|
from app.models.domain.users import User
|
|
from app.models.schemas.articles import (
|
|
DEFAULT_ARTICLES_LIMIT,
|
|
DEFAULT_ARTICLES_OFFSET,
|
|
ArticlesFilters,
|
|
)
|
|
from app.resources import strings
|
|
from app.services.articles import check_user_can_modify_article
|
|
|
|
|
|
def get_articles_filters(
|
|
tag: Optional[str] = None,
|
|
tags: Optional[List[str]] = Query(default=None),
|
|
author: Optional[str] = None,
|
|
favorited: Optional[str] = None,
|
|
search: Optional[str] = Query(default=None, description="搜索标题/描述/slug"),
|
|
limit: int = Query(DEFAULT_ARTICLES_LIMIT, ge=1),
|
|
offset: int = Query(DEFAULT_ARTICLES_OFFSET, ge=0),
|
|
) -> ArticlesFilters:
|
|
final_tags: Optional[List[str]] = tags
|
|
# 兼容旧的 tag 单值参数：若提供则与 tags 合并并去重
|
|
if tag:
|
|
final_tags = list({*(final_tags or []), tag})
|
|
return ArticlesFilters(
|
|
tag=tag,
|
|
tags=final_tags,
|
|
author=author,
|
|
favorited=favorited,
|
|
search=search,
|
|
limit=limit,
|
|
offset=offset,
|
|
)
|
|
|
|
|
|
async def get_article_by_slug_from_path(
|
|
slug: str = Path(..., min_length=1),
|
|
user: Optional[User] = Depends(get_current_user_authorizer(required=False)),
|
|
articles_repo: ArticlesRepository = Depends(get_repository(ArticlesRepository)),
|
|
) -> Article:
|
|
try:
|
|
return await articles_repo.get_article_by_slug(slug=slug, requested_user=user)
|
|
except EntityDoesNotExist:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=strings.ARTICLE_DOES_NOT_EXIST_ERROR,
|
|
)
|
|
|
|
|
|
def check_article_modification_permissions(
|
|
current_article: Article = Depends(get_article_by_slug_from_path),
|
|
user: User = Depends(get_current_user_authorizer()),
|
|
) -> None:
|
|
if not check_user_can_modify_article(current_article, user):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=strings.USER_IS_NOT_AUTHOR_OF_ARTICLE,
|
|
)
|