Sergey Vanyushkin
3cf6c94da2
- PostLike domain entity (post_id, liked_by) with BaseEntity integration
- Post entity: add like_count field (default 0) and to_dict serialization
- PostRepository interface: add get_like, add_like, remove_like methods
- TogglePostLikeUseCase: toggle logic (like → unlike, unlike → like)
- PostResponseDTO/PostResponseSchema: add like_count field
- PostLikeORM model with FK to posts and cascade delete
- SQLAlchemyPostRepository: implement like query/add/remove with ORM mapping
- DI provider registration for TogglePostLikeUseCase
- API endpoint POST /api/v1/posts/{id}/like (auth required)
- Unit tests: PostLike entity, Post.like_count, TogglePostLikeUseCase (7 tests)
- API tests: POST /api/v1/posts/{id}/like (4 tests)
- Test model files: FEATURE_LIKES.md, TEST_MODEL.md updated
375 lines
9.5 KiB
Python
375 lines
9.5 KiB
Python
"""Posts API routes.
|
|
|
|
This module defines FastAPI routes for blog post operations.
|
|
Implements CRUD endpoints with authentication and authorization.
|
|
"""
|
|
|
|
from uuid import UUID
|
|
|
|
from dishka.integrations.fastapi import DishkaRoute
|
|
from fastapi import APIRouter, status
|
|
|
|
from app.application.dtos import CreatePostDTO, UpdatePostDTO
|
|
from app.domain.exceptions import ForbiddenException
|
|
from app.domain.roles import Permission, has_permission
|
|
from app.presentation.api.deps import (
|
|
CreatePostDep,
|
|
CurrentRoleDep,
|
|
CurrentUserDep,
|
|
DeletePostDep,
|
|
GetPostDep,
|
|
ListPostsDep,
|
|
PublishPostDep,
|
|
ToggleLikeDep,
|
|
UpdatePostDep,
|
|
)
|
|
from app.presentation.schemas import (
|
|
PostCreateSchema,
|
|
PostListResponseSchema,
|
|
PostResponseSchema,
|
|
PostUpdateSchema,
|
|
)
|
|
|
|
router = APIRouter(prefix="/posts", tags=["posts"], route_class=DishkaRoute)
|
|
|
|
|
|
@router.post(
|
|
"",
|
|
response_model=PostResponseSchema,
|
|
status_code=status.HTTP_201_CREATED,
|
|
summary="Create a new post",
|
|
)
|
|
async def create_post(
|
|
schema: PostCreateSchema,
|
|
use_case: CreatePostDep,
|
|
current_user_id: CurrentUserDep,
|
|
) -> PostResponseSchema:
|
|
"""Create a new blog post.
|
|
|
|
Args:
|
|
schema: Post creation data.
|
|
use_case: CreatePostUseCase dependency.
|
|
current_user_id: Authenticated user ID.
|
|
|
|
Returns:
|
|
PostResponseSchema with created post data.
|
|
"""
|
|
dto = CreatePostDTO(
|
|
title=schema.title,
|
|
content=schema.content,
|
|
author_id=current_user_id,
|
|
tags=schema.tags,
|
|
)
|
|
result = await use_case.execute(dto)
|
|
return PostResponseSchema(**result.__dict__)
|
|
|
|
|
|
@router.get(
|
|
"",
|
|
response_model=PostListResponseSchema,
|
|
summary="List posts",
|
|
)
|
|
async def list_posts(
|
|
use_case: ListPostsDep,
|
|
role: CurrentRoleDep,
|
|
include_unpublished: bool = False,
|
|
limit: int = 10,
|
|
offset: int = 0,
|
|
) -> PostListResponseSchema:
|
|
"""Get blog posts with optional filtering and pagination.
|
|
|
|
Args:
|
|
use_case: ListPostsUseCase dependency.
|
|
role: Current user role.
|
|
include_unpublished: If True, returns all posts including drafts.
|
|
Only admins can use this parameter.
|
|
limit: Maximum number of posts to return (default: 10, max: 100).
|
|
offset: Number of posts to skip (default: 0).
|
|
|
|
Returns:
|
|
PostListResponseSchema with paginated posts.
|
|
|
|
Raises:
|
|
ForbiddenException: If non-admin tries to include unpublished posts.
|
|
"""
|
|
limit = max(1, min(limit, 100))
|
|
offset = max(0, offset)
|
|
|
|
if include_unpublished:
|
|
if not has_permission(role, Permission.POST_READ_UNPUBLISHED):
|
|
raise ForbiddenException("Only admins can view unpublished posts")
|
|
results = await use_case.all_posts()
|
|
else:
|
|
results = await use_case.published_posts(limit=limit, offset=offset)
|
|
|
|
items = [PostResponseSchema(**r.__dict__) for r in results]
|
|
return PostListResponseSchema(items=items, total=len(items))
|
|
|
|
|
|
@router.get(
|
|
"/published",
|
|
response_model=PostListResponseSchema,
|
|
summary="List published posts",
|
|
)
|
|
async def list_published_posts(
|
|
use_case: ListPostsDep,
|
|
) -> PostListResponseSchema:
|
|
"""Get all published blog posts.
|
|
|
|
Args:
|
|
use_case: ListPostsUseCase dependency.
|
|
|
|
Returns:
|
|
PostListResponseSchema with published posts.
|
|
"""
|
|
results = await use_case.published_posts()
|
|
items = [PostResponseSchema(**r.__dict__) for r in results]
|
|
return PostListResponseSchema(items=items, total=len(items))
|
|
|
|
|
|
@router.get(
|
|
"/search",
|
|
response_model=PostListResponseSchema,
|
|
summary="Search posts",
|
|
)
|
|
async def search_posts(
|
|
query: str,
|
|
use_case: ListPostsDep,
|
|
) -> PostListResponseSchema:
|
|
"""Search posts by query.
|
|
|
|
Args:
|
|
query: Search query string.
|
|
use_case: ListPostsUseCase dependency.
|
|
|
|
Returns:
|
|
PostListResponseSchema with matching posts.
|
|
"""
|
|
results = await use_case.search(query)
|
|
items = [PostResponseSchema(**r.__dict__) for r in results]
|
|
return PostListResponseSchema(items=items, total=len(items))
|
|
|
|
|
|
@router.get(
|
|
"/by-tag/{tag}",
|
|
response_model=PostListResponseSchema,
|
|
summary="Get posts by tag",
|
|
)
|
|
async def get_posts_by_tag(
|
|
tag: str,
|
|
use_case: ListPostsDep,
|
|
) -> PostListResponseSchema:
|
|
"""Get posts by tag.
|
|
|
|
Args:
|
|
tag: Tag to filter by.
|
|
use_case: ListPostsUseCase dependency.
|
|
|
|
Returns:
|
|
PostListResponseSchema with tagged posts.
|
|
"""
|
|
results = await use_case.by_tag(tag)
|
|
items = [PostResponseSchema(**r.__dict__) for r in results]
|
|
return PostListResponseSchema(items=items, total=len(items))
|
|
|
|
|
|
@router.get(
|
|
"/by-author/{author_id}",
|
|
response_model=PostListResponseSchema,
|
|
summary="Get posts by author",
|
|
)
|
|
async def get_posts_by_author(
|
|
author_id: str,
|
|
use_case: ListPostsDep,
|
|
) -> PostListResponseSchema:
|
|
"""Get posts by author.
|
|
|
|
Args:
|
|
author_id: Author identifier.
|
|
use_case: ListPostsUseCase dependency.
|
|
|
|
Returns:
|
|
PostListResponseSchema with author's posts.
|
|
"""
|
|
results = await use_case.by_author(author_id)
|
|
items = [PostResponseSchema(**r.__dict__) for r in results]
|
|
return PostListResponseSchema(items=items, total=len(items))
|
|
|
|
|
|
@router.get(
|
|
"/{post_id}",
|
|
response_model=PostResponseSchema,
|
|
summary="Get post by ID",
|
|
)
|
|
async def get_post(
|
|
post_id: UUID,
|
|
use_case: GetPostDep,
|
|
) -> PostResponseSchema:
|
|
"""Get a post by its ID.
|
|
|
|
Args:
|
|
post_id: Unique post identifier.
|
|
use_case: GetPostUseCase dependency.
|
|
|
|
Returns:
|
|
PostResponseSchema with post data.
|
|
"""
|
|
result = await use_case.by_id(post_id)
|
|
return PostResponseSchema(**result.__dict__)
|
|
|
|
|
|
@router.get(
|
|
"/slug/{slug}",
|
|
response_model=PostResponseSchema,
|
|
summary="Get post by slug",
|
|
)
|
|
async def get_post_by_slug(
|
|
slug: str,
|
|
use_case: GetPostDep,
|
|
) -> PostResponseSchema:
|
|
"""Get a post by its slug.
|
|
|
|
Args:
|
|
slug: URL-friendly slug identifier.
|
|
use_case: GetPostUseCase dependency.
|
|
|
|
Returns:
|
|
PostResponseSchema with post data.
|
|
"""
|
|
result = await use_case.by_slug(slug)
|
|
return PostResponseSchema(**result.__dict__)
|
|
|
|
|
|
@router.patch(
|
|
"/{post_id}",
|
|
response_model=PostResponseSchema,
|
|
summary="Update post",
|
|
)
|
|
async def update_post(
|
|
post_id: UUID,
|
|
schema: PostUpdateSchema,
|
|
use_case: UpdatePostDep,
|
|
current_user_id: CurrentUserDep,
|
|
role: CurrentRoleDep,
|
|
) -> PostResponseSchema:
|
|
"""Update a post.
|
|
|
|
Args:
|
|
post_id: Unique post identifier.
|
|
schema: Update data.
|
|
use_case: UpdatePostUseCase dependency.
|
|
current_user_id: Authenticated user ID.
|
|
role: Current user role.
|
|
|
|
Returns:
|
|
PostResponseSchema with updated post data.
|
|
"""
|
|
dto = UpdatePostDTO(
|
|
title=schema.title,
|
|
content=schema.content,
|
|
tags=schema.tags,
|
|
)
|
|
result = await use_case.execute(post_id, dto, current_user_id, role)
|
|
return PostResponseSchema(**result.__dict__)
|
|
|
|
|
|
@router.delete(
|
|
"/{post_id}",
|
|
status_code=status.HTTP_204_NO_CONTENT,
|
|
summary="Delete post",
|
|
)
|
|
async def delete_post(
|
|
post_id: UUID,
|
|
use_case: DeletePostDep,
|
|
current_user_id: CurrentUserDep,
|
|
role: CurrentRoleDep,
|
|
) -> None:
|
|
"""Delete a post.
|
|
|
|
Args:
|
|
post_id: Unique post identifier.
|
|
use_case: DeletePostUseCase dependency.
|
|
current_user_id: Authenticated user ID.
|
|
role: Current user role.
|
|
"""
|
|
await use_case.execute(post_id, current_user_id, role)
|
|
|
|
|
|
@router.post(
|
|
"/{post_id}/publish",
|
|
response_model=PostResponseSchema,
|
|
summary="Publish post",
|
|
)
|
|
async def publish_post(
|
|
post_id: UUID,
|
|
use_case: PublishPostDep,
|
|
current_user_id: CurrentUserDep,
|
|
role: CurrentRoleDep,
|
|
) -> PostResponseSchema:
|
|
"""Publish a post.
|
|
|
|
Args:
|
|
post_id: Unique post identifier.
|
|
use_case: PublishPostUseCase dependency.
|
|
current_user_id: Authenticated user ID.
|
|
role: Current user role.
|
|
|
|
Returns:
|
|
PostResponseSchema with published post data.
|
|
"""
|
|
result = await use_case.publish(post_id, current_user_id, role)
|
|
return PostResponseSchema(**result.__dict__)
|
|
|
|
|
|
@router.post(
|
|
"/{post_id}/unpublish",
|
|
response_model=PostResponseSchema,
|
|
summary="Unpublish post",
|
|
)
|
|
async def unpublish_post(
|
|
post_id: UUID,
|
|
use_case: PublishPostDep,
|
|
current_user_id: CurrentUserDep,
|
|
role: CurrentRoleDep,
|
|
) -> PostResponseSchema:
|
|
"""Unpublish a post.
|
|
|
|
Args:
|
|
post_id: Unique post identifier.
|
|
use_case: PublishPostUseCase dependency.
|
|
current_user_id: Authenticated user ID.
|
|
role: Current user role.
|
|
|
|
Returns:
|
|
PostResponseSchema with unpublished post data.
|
|
"""
|
|
result = await use_case.unpublish(post_id, current_user_id, role)
|
|
return PostResponseSchema(**result.__dict__)
|
|
|
|
|
|
@router.post(
|
|
"/{post_id}/like",
|
|
response_model=PostResponseSchema,
|
|
summary="Toggle like on a post",
|
|
)
|
|
async def toggle_like(
|
|
post_id: UUID,
|
|
use_case: ToggleLikeDep,
|
|
current_user_id: CurrentUserDep,
|
|
) -> PostResponseSchema:
|
|
"""Toggle like/unlike on a post.
|
|
|
|
If the user already liked the post, the like is removed (unlike).
|
|
Otherwise, a new like is added.
|
|
|
|
Args:
|
|
post_id: Unique identifier of the post.
|
|
use_case: TogglePostLikeUseCase dependency.
|
|
current_user_id: Authenticated user ID.
|
|
|
|
Returns:
|
|
PostResponseSchema with updated like_count.
|
|
"""
|
|
result = await use_case.execute(post_id, current_user_id)
|
|
return PostResponseSchema(**result.__dict__)
|