Felipe Cardoso
270bfda1f8
Introduce a new API for managing event guests, including endpoints for creating, reading, updating, deleting, and changing guest status. Added corresponding Pydantic schemas, database CRUD logic, and extensive test coverage to validate functionality. Integrated the guests API under the events router.
46 lines
1.6 KiB
Python
46 lines
1.6 KiB
Python
from app.crud.base import CRUDBase
|
|
from app.models.guest import Guest
|
|
from app.schemas.guests import GuestCreate, GuestUpdate
|
|
from sqlalchemy.orm import Session
|
|
from typing import Optional
|
|
from app.models.guest import GuestStatus
|
|
from datetime import datetime, timezone
|
|
import uuid
|
|
|
|
|
|
class CRUDGuest(CRUDBase[Guest, GuestCreate, GuestUpdate]):
|
|
|
|
def create(self, db, obj_in: GuestCreate):
|
|
db_guest = Guest(
|
|
event_id=uuid.UUID(obj_in.event_id) if isinstance(obj_in.event_id, str) else obj_in.event_id, # explicit casting
|
|
invited_by=uuid.UUID(obj_in.invited_by) if isinstance(obj_in.invited_by, str) else obj_in.invited_by,
|
|
full_name=obj_in.full_name,
|
|
email=obj_in.email,
|
|
invitation_code=obj_in.invitation_code
|
|
)
|
|
db.add(db_guest)
|
|
db.commit()
|
|
db.refresh(db_guest)
|
|
return db_guest
|
|
|
|
|
|
def get_by_invitation_code(self, db: Session, invitation_code: str) -> Optional[Guest]:
|
|
return db.query(Guest).filter(Guest.invitation_code == invitation_code).first()
|
|
|
|
def update_status(self, db: Session, guest_id: uuid.UUID, status: GuestStatus):
|
|
guest = self.get(db, guest_id)
|
|
if guest:
|
|
guest.status = status
|
|
guest.response_date = datetime.now(timezone.utc)
|
|
db.commit()
|
|
db.refresh(guest)
|
|
return guest
|
|
|
|
def remove(self, db: Session, *, guest_id: str) -> Guest:
|
|
guest_id = uuid.UUID(guest_id) if isinstance(guest_id, str) else guest_id
|
|
obj = db.query(self.model).get(guest_id)
|
|
db.delete(obj)
|
|
db.commit()
|
|
return obj
|
|
|
|
guest_crud = CRUDGuest(Guest) |