agatha
2db6e18d5b
Reviewed-on: #2 Co-authored-by: agatha <agatha@juggalol.com> Co-committed-by: agatha <agatha@juggalol.com>
107 lines
2.6 KiB
Python
107 lines
2.6 KiB
Python
from typing import Annotated
|
|
from fastapi import APIRouter, Depends, HTTPException, Path
|
|
from sqlalchemy.orm import Session
|
|
from pydantic import BaseModel, Field
|
|
from starlette import status
|
|
|
|
from database import SessionLocal
|
|
from models import Post, Thread
|
|
|
|
router = APIRouter(
|
|
tags=['forum']
|
|
)
|
|
|
|
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
db_dependency = Annotated[Session, Depends(get_db)]
|
|
|
|
|
|
class PostCreate(BaseModel):
|
|
author: str = Field('anon')
|
|
title: str = Field('')
|
|
content: str = Field('')
|
|
|
|
|
|
@router.get('/catalog', status_code=status.HTTP_200_OK)
|
|
async def get_catalog(db: db_dependency):
|
|
return db.query(Thread).all()
|
|
|
|
|
|
@router.get('/', status_code=status.HTTP_200_OK)
|
|
async def get_posts(db: db_dependency):
|
|
return db.query(Post).all()
|
|
|
|
|
|
@router.get('/{thread_id}', status_code=status.HTTP_200_OK)
|
|
async def get_thread_by_id(db: db_dependency, thread_id: int = Path(gt=0)):
|
|
posts = db.query(Post).filter(Post.thread_id == thread_id).all()
|
|
if posts:
|
|
return posts
|
|
|
|
raise HTTPException(404, f'Could not find thread')
|
|
|
|
@router.post('/', status_code=status.HTTP_201_CREATED)
|
|
async def create_thread(db: db_dependency, data: PostCreate):
|
|
try:
|
|
# Create the post
|
|
post = Post(
|
|
author=data.author,
|
|
title=data.title,
|
|
content=data.content
|
|
)
|
|
db.add(post)
|
|
db.flush()
|
|
|
|
# Create the thread
|
|
thread = Thread(
|
|
author=post.author,
|
|
title=post.title,
|
|
content=post.content
|
|
)
|
|
db.add(thread)
|
|
db.flush()
|
|
|
|
# Update the Post with thread_id
|
|
post.thread_id = thread.id
|
|
db.commit()
|
|
|
|
return {
|
|
'id': post.id,
|
|
'author': post.author,
|
|
'title': post.title,
|
|
'content': post.content
|
|
}
|
|
except Exception as e:
|
|
db.rollback()
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
|
|
@router.post('/{thread_id}', status_code=status.HTTP_201_CREATED)
|
|
async def create_reply(db: db_dependency, data: PostCreate, thread_id: int = Path(gt=0)):
|
|
thread = db.query(Thread).filter(Thread.id == thread_id).first()
|
|
if thread:
|
|
post = Post(
|
|
thread_id=thread_id,
|
|
author=data.author,
|
|
title=data.title,
|
|
content=data.content
|
|
)
|
|
db.add(post)
|
|
db.commit()
|
|
|
|
return {
|
|
'id': post.id,
|
|
'author': post.author,
|
|
'title': post.title,
|
|
'content': post.content
|
|
}
|
|
|
|
raise HTTPException(status_code=404, detail='Could not find thread')
|