606cda0912
- Create project structure with app organization - Set up FastAPI application with CORS and health endpoint - Implement database models with SQLAlchemy (User, Post, Comment) - Set up Alembic for database migrations - Implement authentication with JWT tokens - Create CRUD operations for all models - Implement REST API endpoints for users, posts, and comments - Add comprehensive documentation in README.md
148 lines
4.2 KiB
Python
148 lines
4.2 KiB
Python
from typing import Any, List
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import crud, models, schemas
|
|
from app.auth import deps
|
|
from app.db.session import get_db
|
|
|
|
router = APIRouter()
|
|
|
|
@router.get("/", response_model=List[schemas.Post])
|
|
def read_posts(
|
|
db: Session = Depends(get_db),
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
) -> Any:
|
|
"""
|
|
Retrieve published posts.
|
|
"""
|
|
posts = crud.post.get_multi_published(db, skip=skip, limit=limit)
|
|
return posts
|
|
|
|
@router.post("/", response_model=schemas.Post)
|
|
def create_post(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
post_in: schemas.PostCreate,
|
|
current_user: models.User = Depends(deps.get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Create new post.
|
|
"""
|
|
post = crud.post.create_with_author(db, obj_in=post_in, author_id=current_user.id)
|
|
return post
|
|
|
|
@router.get("/{post_id}", response_model=schemas.Post)
|
|
def read_post(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
post_id: str,
|
|
) -> Any:
|
|
"""
|
|
Get post by ID.
|
|
"""
|
|
post = crud.post.get(db, id=post_id)
|
|
if not post:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Post not found",
|
|
)
|
|
if not post.is_published:
|
|
# Only the author can see unpublished posts
|
|
current_user = deps.get_current_user(db=db)
|
|
if current_user.id != post.author_id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="The post is not published yet",
|
|
)
|
|
return post
|
|
|
|
@router.put("/{post_id}", response_model=schemas.Post)
|
|
def update_post(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
post_id: str,
|
|
post_in: schemas.PostUpdate,
|
|
current_user: models.User = Depends(deps.get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Update a post.
|
|
"""
|
|
post = crud.post.get(db, id=post_id)
|
|
if not post:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Post not found",
|
|
)
|
|
if post.author_id != current_user.id and not current_user.is_superuser:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="You can only update your own posts",
|
|
)
|
|
post = crud.post.update(db, db_obj=post, obj_in=post_in)
|
|
return post
|
|
|
|
@router.delete("/{post_id}", response_model=schemas.Post)
|
|
def delete_post(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
post_id: str,
|
|
current_user: models.User = Depends(deps.get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Delete a post.
|
|
"""
|
|
post = crud.post.get(db, id=post_id)
|
|
if not post:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Post not found",
|
|
)
|
|
if post.author_id != current_user.id and not current_user.is_superuser:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="You can only delete your own posts",
|
|
)
|
|
post = crud.post.remove(db, id=post_id)
|
|
return post
|
|
|
|
@router.get("/user/{user_id}", response_model=List[schemas.Post])
|
|
def read_user_posts(
|
|
user_id: str,
|
|
db: Session = Depends(get_db),
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
) -> Any:
|
|
"""
|
|
Retrieve posts by user.
|
|
"""
|
|
user = crud.user.get(db, id=user_id)
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="User not found",
|
|
)
|
|
|
|
# Check if the current user is the author or a superuser
|
|
try:
|
|
current_user = deps.get_current_user(db=db)
|
|
is_author_or_superuser = (user_id == current_user.id) or current_user.is_superuser
|
|
except Exception:
|
|
is_author_or_superuser = False
|
|
|
|
if is_author_or_superuser:
|
|
# Return all posts for the author or superuser
|
|
posts = crud.post.get_multi_by_author(
|
|
db=db, author_id=user_id, skip=skip, limit=limit
|
|
)
|
|
else:
|
|
# Filter to only published posts for other users
|
|
posts = [
|
|
post for post in crud.post.get_multi_by_author(
|
|
db=db, author_id=user_id, skip=skip, limit=limit
|
|
)
|
|
if post.is_published
|
|
]
|
|
return posts |