Add event routes and enhance routing structure

Introduced routes for event management, including CRUD operations and querying by user or public visibility. Updated event themes routes with operation IDs for better documentation. Refactored `api/main.py` to `api/router.py` and integrated events routing into the API.
This commit is contained in:
2025-03-05 18:57:24 +01:00
parent 4a9a37f507
commit 5c73f2720e
4 changed files with 149 additions and 5 deletions

View File

@@ -0,0 +1,141 @@
from datetime import datetime
from typing import List, Optional
from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from api.dependencies.auth import get_current_user
from app.core.database import get_db
from app.crud.event import event
from app.models.user import User
from app.schemas.events import (
EventCreate,
EventUpdate,
EventResponse
)
events_router = APIRouter()
@events_router.post("/", response_model=EventResponse, operation_id="create_event")
def create_event(
*,
db: Session = Depends(get_db),
event_in: EventCreate,
current_user: User = Depends(get_current_user)
) -> EventResponse:
"""Create a new event."""
# Check if slug is already taken
if event.get_by_slug(db, slug=event_in.slug):
raise HTTPException(
status_code=400,
detail="An event with this slug already exists"
)
return event.create_with_owner(
db=db,
obj_in=event_in,
owner_id=current_user.id
)
@events_router.get("/me", response_model=List[EventResponse], operation_id="get_user_events")
def get_user_events(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
include_inactive: bool = False,
current_user: User = Depends(get_current_user)
) -> List[EventResponse]:
"""Get all events created by the current user."""
return event.get_user_events(
db=db,
user_id=current_user.id,
skip=skip,
limit=limit,
include_inactive=include_inactive
)
@events_router.get("/upcoming", response_model=List[EventResponse], operation_id="get_upcoming_events")
def get_upcoming_events(
*,
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100
) -> List[EventResponse]:
"""Get upcoming public events."""
return event.get_upcoming_events(db=db, skip=skip, limit=limit)
@events_router.get("/{event_id}", response_model=EventResponse, operation_id="get_event")
def get_event(
*,
db: Session = Depends(get_db),
event_id: UUID,
current_user: User = Depends(get_current_user)
) -> EventResponse:
"""Get event by ID."""
event_obj = event.get(db=db, id=event_id)
if not event_obj:
raise HTTPException(status_code=404, detail="Event not found")
return event_obj
@events_router.get("/by-slug/{slug}", response_model=EventResponse, operation_id="get_public_event")
def get_public_event(
*,
db: Session = Depends(get_db),
slug: str,
access_code: Optional[str] = Query(None)
) -> EventResponse:
"""Get public event by slug."""
event_obj = event.get_public_event(db=db, slug=slug, access_code=access_code)
if not event_obj:
raise HTTPException(status_code=404, detail="Event not found")
return event_obj
@events_router.put("/{event_id}", response_model=EventResponse, operation_id="update_event")
def update_event(
*,
db: Session = Depends(get_db),
event_id: UUID,
event_in: EventUpdate,
current_user: User = Depends(get_current_user)
) -> EventResponse:
"""Update event."""
event_obj = event.get(db=db, id=event_id)
if not event_obj:
raise HTTPException(status_code=404, detail="Event not found")
if event_obj.created_by != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
# If slug is being updated, check if new slug is available
if event_in.slug and event_in.slug != event_obj.slug:
if event.get_by_slug(db, slug=event_in.slug):
raise HTTPException(
status_code=400,
detail="An event with this slug already exists"
)
return event.update(db=db, db_obj=event_obj, obj_in=event_in)
@events_router.delete("/{event_id}", operation_id="delete_event")
def delete_event(
*,
db: Session = Depends(get_db),
event_id: UUID,
current_user: User = Depends(get_current_user)
):
"""Delete event."""
event_obj = event.get(db=db, id=event_id)
if not event_obj:
raise HTTPException(status_code=404, detail="Event not found")
if event_obj.created_by != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
event.remove(db=db, id=event_id)
return {"message": "Event deleted successfully"}