06df0285b1
- Set up project structure and dependencies - Create database models for users, posts, comments, and tags - Set up Alembic for database migrations - Implement user authentication (register, login) - Create CRUD endpoints for blog posts, comments, and tags - Add health check endpoint - Set up proper error handling - Update README with project details and setup instructions
80 lines
2.3 KiB
Python
80 lines
2.3 KiB
Python
from typing import List, Optional
|
|
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
from app.crud.base import CRUDBase
|
|
from app.models.post import Post
|
|
from app.models.tag import Tag
|
|
from app.schemas.post import PostCreate, PostUpdate
|
|
|
|
|
|
class CRUDPost(CRUDBase[Post, PostCreate, PostUpdate]):
|
|
def create_with_author(
|
|
self, db: Session, *, obj_in: PostCreate, author_id: int
|
|
) -> Post:
|
|
obj_in_data = obj_in.model_dump()
|
|
tag_ids = obj_in_data.pop("tag_ids", [])
|
|
db_obj = Post(**obj_in_data, author_id=author_id)
|
|
|
|
# Add tags if provided
|
|
if tag_ids:
|
|
tags = db.query(Tag).filter(Tag.id.in_(tag_ids)).all()
|
|
db_obj.tags = tags
|
|
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def get_multi_by_author(
|
|
self, db: Session, *, author_id: int, skip: int = 0, limit: int = 100
|
|
) -> List[Post]:
|
|
return (
|
|
db.query(Post)
|
|
.filter(Post.author_id == author_id)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def get_with_details(self, db: Session, *, id: int) -> Optional[Post]:
|
|
return (
|
|
db.query(Post)
|
|
.options(joinedload(Post.author), joinedload(Post.tags))
|
|
.filter(Post.id == id)
|
|
.first()
|
|
)
|
|
|
|
def get_multi_with_details(
|
|
self, db: Session, *, skip: int = 0, limit: int = 100
|
|
) -> List[Post]:
|
|
return (
|
|
db.query(Post)
|
|
.options(joinedload(Post.author), joinedload(Post.tags))
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def update_with_tags(
|
|
self, db: Session, *, db_obj: Post, obj_in: PostUpdate
|
|
) -> Post:
|
|
obj_in_data = obj_in.model_dump(exclude_unset=True)
|
|
tag_ids = obj_in_data.pop("tag_ids", None)
|
|
|
|
# Update basic fields
|
|
for field in obj_in_data:
|
|
setattr(db_obj, field, obj_in_data[field])
|
|
|
|
# Update tags if provided
|
|
if tag_ids is not None:
|
|
tags = db.query(Tag).filter(Tag.id.in_(tag_ids)).all()
|
|
db_obj.tags = tags
|
|
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
|
|
post = CRUDPost(Post) |