from typing import List, Optional from sqlalchemy.orm import Session from sqlalchemy import exc from models.person import Person from schemas.person import PersonCreate, PersonUpdate, PersonSchema import uuid def get_all_people(db: Session) -> List[PersonSchema]: """ Retrieves all people from the database. Args: db (Session): The database session. Returns: List[PersonSchema]: A list of all person objects. """ people = db.query(Person).all() return [PersonSchema.from_orm(person) for person in people] def get_person_by_id(db: Session, person_id: uuid.UUID) -> Optional[PersonSchema]: """ Retrieves a single person by their ID. Args: db (Session): The database session. person_id (UUID): The ID of the person to retrieve. Returns: Optional[PersonSchema]: The person object if found, otherwise None. """ person = db.query(Person).filter(Person.id == person_id).first() return PersonSchema.from_orm(person) if person else None def create_person(db: Session, person_data: PersonCreate) -> PersonSchema: """ Creates a new person in the database. Args: db (Session): The database session. person_data (PersonCreate): The data for the person to create. Returns: PersonSchema: The newly created person object. """ try: db_person = Person(**person_data.dict()) db.add(db_person) db.commit() db.refresh(db_person) return PersonSchema.from_orm(db_person) except exc.IntegrityError as e: db.rollback() raise ValueError(f"Error creating person: {str(e)}") def update_person(db: Session, person_id: uuid.UUID, person_data: PersonUpdate) -> Optional[PersonSchema]: """ Updates an existing person in the database. Args: db (Session): The database session. person_id (UUID): The ID of the person to update. person_data (PersonUpdate): The updated data for the person. Returns: Optional[PersonSchema]: The updated person object if found, otherwise None. """ person = db.query(Person).filter(Person.id == person_id).first() if person: update_data = person_data.dict(exclude_unset=True) for key, value in update_data.items(): setattr(person, key, value) db.commit() db.refresh(person) return PersonSchema.from_orm(person) return None def delete_person(db: Session, person_id: uuid.UUID) -> bool: """ Deletes a person from the database. Args: db (Session): The database session. person_id (UUID): The ID of the person to delete. Returns: bool: True if the person was deleted, False otherwise. """ person = db.query(Person).filter(Person.id == person_id).first() if person: db.delete(person) db.commit() return True return False