6d3b1188d1
- Setup project structure with FastAPI - Create database models for users, gifts, preferences, and recommendations - Configure SQLite database with SQLAlchemy ORM - Setup Alembic for database migrations - Implement user authentication with JWT - Create API endpoints for users, gifts, preferences, and recommendations - Integrate OpenAI API for gift recommendations - Add comprehensive documentation
138 lines
3.9 KiB
Python
138 lines
3.9 KiB
Python
from typing import Any, List
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Path, Query
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db.deps import get_current_active_user, get_db
|
|
from app.models.preference import Preference
|
|
from app.models.user import User
|
|
from app.schemas.preference import Preference as PreferenceSchema
|
|
from app.schemas.preference import PreferenceCreate, PreferenceUpdate
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("", response_model=List[PreferenceSchema])
|
|
def read_preferences(
|
|
db: Session = Depends(get_db),
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
recipient_name: str = Query(None, description="Filter by recipient name"),
|
|
current_user: User = Depends(get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Retrieve preferences.
|
|
"""
|
|
query = db.query(Preference).filter(Preference.user_id == current_user.id)
|
|
|
|
# Apply filters if provided
|
|
if recipient_name:
|
|
query = query.filter(Preference.recipient_name == recipient_name)
|
|
|
|
# Apply pagination
|
|
preferences = query.offset(skip).limit(limit).all()
|
|
return preferences
|
|
|
|
|
|
@router.post("", response_model=PreferenceSchema)
|
|
def create_preference(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
preference_in: PreferenceCreate,
|
|
current_user: User = Depends(get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Create new preference.
|
|
"""
|
|
# Check if a preference for this recipient already exists
|
|
existing = db.query(Preference).filter(
|
|
Preference.user_id == current_user.id,
|
|
Preference.recipient_name == preference_in.recipient_name
|
|
).first()
|
|
|
|
if existing:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Preference for recipient '{preference_in.recipient_name}' already exists. Use PUT to update."
|
|
)
|
|
|
|
preference = Preference(
|
|
**preference_in.dict(),
|
|
user_id=current_user.id
|
|
)
|
|
db.add(preference)
|
|
db.commit()
|
|
db.refresh(preference)
|
|
return preference
|
|
|
|
|
|
@router.put("/{id}", response_model=PreferenceSchema)
|
|
def update_preference(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
id: int = Path(..., description="The ID of the preference to update"),
|
|
preference_in: PreferenceUpdate,
|
|
current_user: User = Depends(get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Update a preference.
|
|
"""
|
|
preference = db.query(Preference).filter(
|
|
Preference.id == id,
|
|
Preference.user_id == current_user.id
|
|
).first()
|
|
|
|
if not preference:
|
|
raise HTTPException(status_code=404, detail="Preference not found")
|
|
|
|
update_data = preference_in.dict(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(preference, field, value)
|
|
|
|
db.add(preference)
|
|
db.commit()
|
|
db.refresh(preference)
|
|
return preference
|
|
|
|
|
|
@router.get("/{id}", response_model=PreferenceSchema)
|
|
def read_preference(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
id: int = Path(..., description="The ID of the preference to get"),
|
|
current_user: User = Depends(get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Get preference by ID.
|
|
"""
|
|
preference = db.query(Preference).filter(
|
|
Preference.id == id,
|
|
Preference.user_id == current_user.id
|
|
).first()
|
|
|
|
if not preference:
|
|
raise HTTPException(status_code=404, detail="Preference not found")
|
|
return preference
|
|
|
|
|
|
@router.delete("/{id}", status_code=204, response_model=None)
|
|
def delete_preference(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
id: int = Path(..., description="The ID of the preference to delete"),
|
|
current_user: User = Depends(get_current_active_user),
|
|
) -> Any:
|
|
"""
|
|
Delete a preference.
|
|
"""
|
|
preference = db.query(Preference).filter(
|
|
Preference.id == id,
|
|
Preference.user_id == current_user.id
|
|
).first()
|
|
|
|
if not preference:
|
|
raise HTTPException(status_code=404, detail="Preference not found")
|
|
|
|
db.delete(preference)
|
|
db.commit()
|
|
return None |