Add helper functions for Dog

helpers/dog_helpers.py

@@ -0,0 +1,116 @@
+from typing import List, Optional
+from pydantic import BaseModel, validator
+from sqlalchemy.orm import Session
+from models import Dog, User
+
+class DogCreate(BaseModel):
+    name: str
+    breed: str
+    age: int
+    owner_id: str
+
+    @validator('name')
+    def validate_name(cls, name):
+        if not name.isalpha():
+            raise ValueError('Name should only contain alphabetic characters')
+        return name
+
+    @validator('breed')
+    def validate_breed(cls, breed):
+        if not breed.isalpha():
+            raise ValueError('Breed should only contain alphabetic characters')
+        return breed
+
+    @validator('age')
+    def validate_age(cls, age):
+        if age < 0:
+            raise ValueError('Age cannot be negative')
+        return age
+
+def get_dogs(db: Session, owner_id: Optional[str] = None) -> List[Dog]:
+    """
+    Get a list of dogs from the database.
+
+    Args:
+        db (Session): SQLAlchemy database session
+        owner_id (str, optional): Filter dogs by owner ID
+
+    Returns:
+        List[Dog]: List of Dog objects
+    """
+    query = db.query(Dog)
+    if owner_id:
+        query = query.filter(Dog.owner_id == owner_id)
+    return query.all()
+
+def create_dog(db: Session, dog_data: DogCreate) -> Dog:
+    """
+    Create a new dog in the database.
+
+    Args:
+        db (Session): SQLAlchemy database session
+        dog_data (DogCreate): Pydantic model with dog data
+
+    Returns:
+        Dog: The newly created Dog object
+    """
+    owner = db.query(User).filter(User.id == dog_data.owner_id).first()
+    if not owner:
+        raise ValueError(f"Owner with ID {dog_data.owner_id} not found")
+
+    dog = Dog(
+        name=dog_data.name,
+        breed=dog_data.breed,
+        age=dog_data.age,
+        owner_id=dog_data.owner_id
+    )
+    db.add(dog)
+    db.commit()
+    db.refresh(dog)
+    return dog
+
+def update_dog(db: Session, dog_id: int, dog_data: DogCreate) -> Dog:
+    """
+    Update an existing dog in the database.
+
+    Args:
+        db (Session): SQLAlchemy database session
+        dog_id (int): ID of the dog to update
+        dog_data (DogCreate): Pydantic model with updated dog data
+
+    Returns:
+        Dog: The updated Dog object
+    """
+    dog = db.query(Dog).filter(Dog.id == dog_id).first()
+    if not dog:
+        raise ValueError(f"Dog with ID {dog_id} not found")
+
+    owner = db.query(User).filter(User.id == dog_data.owner_id).first()
+    if not owner:
+        raise ValueError(f"Owner with ID {dog_data.owner_id} not found")
+
+    dog.name = dog_data.name
+    dog.breed = dog_data.breed
+    dog.age = dog_data.age
+    dog.owner_id = dog_data.owner_id
+    db.commit()
+    db.refresh(dog)
+    return dog
+
+def delete_dog(db: Session, dog_id: int) -> None:
+    """
+    Delete a dog from the database.
+
+    Args:
+        db (Session): SQLAlchemy database session
+        dog_id (int): ID of the dog to delete
+
+    Returns:
+        None
+    """
+    dog = db.query(Dog).filter(Dog.id == dog_id).first()
+    if not dog:
+        raise ValueError(f"Dog with ID {dog_id} not found")
+
+    db.delete(dog)
+    db.commit()