from datetime import datetime
from typing import List
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlmodel import Session, select
from app.database import get_session
from app.models import Post, PostCreate, PostRead, PostUpdate, User
router = APIRouter(prefix="/posts", tags=["posts"])
@router.post("/", response_model=PostRead, status_code=status.HTTP_201_CREATED)
def create_post(post: PostCreate, author_id: int, session: Session = Depends(get_session)):
"""Create a new post."""
# Verify author exists
author = session.get(User, author_id)
if not author:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Author not found"
)
# Create new post
db_post = Post(
title=post.title,
content=post.content,
is_published=post.is_published,
author_id=author_id,
published_at=datetime.utcnow() if post.is_published else None
)
session.add(db_post)
session.commit()
session.refresh(db_post)
return db_post
@router.get("/", response_model=List[PostRead])
def read_posts(
skip: int = 0,
limit: int = 100,
published_only: bool = Query(False, description="Filter to published posts only"),
author_id: int = Query(None, description="Filter by author ID"),
session: Session = Depends(get_session)
):
"""Get all posts with optional filters."""
query = select(Post)
if published_only:
query = query.where(Post.is_published == True)
if author_id:
query = query.where(Post.author_id == author_id)
posts = session.exec(query.offset(skip).limit(limit)).all()
return posts
@router.get("/{post_id}", response_model=PostRead)
def read_post(post_id: int, session: Session = Depends(get_session)):
"""Get post by ID."""
post = session.get(Post, post_id)
if not post:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Post not found"
)
return post
@router.put("/{post_id}", response_model=PostRead)
def update_post(post_id: int, post_update: PostUpdate, session: Session = Depends(get_session)):
"""Update post by ID."""
post = session.get(Post, post_id)
if not post:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Post not found"
)
# Update only provided fields
post_data = post_update.model_dump(exclude_unset=True)
# Handle publication status change
if "is_published" in post_data:
if post_data["is_published"] and not post.is_published:
# Publishing for the first time
post.published_at = datetime.utcnow()
elif not post_data["is_published"] and post.is_published:
# Unpublishing
post.published_at = None
# Update other fields
for field, value in post_data.items():
if field != "is_published" or field in post_data:
setattr(post, field, value)
post.updated_at = datetime.utcnow()
session.add(post)
session.commit()
session.refresh(post)
return post
@router.delete("/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_post(post_id: int, session: Session = Depends(get_session)):
"""Delete post by ID."""
post = session.get(Post, post_id)
if not post:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Post not found"
)
session.delete(post)
session.commit()
return None
@router.post("/{post_id}/publish", response_model=PostRead)
def publish_post(post_id: int, session: Session = Depends(get_session)):
"""Publish a post."""
post = session.get(Post, post_id)
if not post:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Post not found"
)
if post.is_published:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Post is already published"
)
post.is_published = True
post.published_at = datetime.utcnow()
post.updated_at = datetime.utcnow()
session.add(post)
session.commit()
session.refresh(post)
return post
@router.post("/{post_id}/unpublish", response_model=PostRead)
def unpublish_post(post_id: int, session: Session = Depends(get_session)):
"""Unpublish a post."""
post = session.get(Post, post_id)
if not post:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Post not found"
)
if not post.is_published:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Post is not published"
)
post.is_published = False
post.published_at = None
post.updated_at = datetime.utcnow()
session.add(post)
session.commit()
session.refresh(post)
return post
