94 lines
2.9 KiB
Python
94 lines
2.9 KiB
Python
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 |