104 lines
2.9 KiB
Python
104 lines
2.9 KiB
Python
from typing import List, Optional
|
|
import uuid
|
|
from sqlalchemy.orm import Session
|
|
from models.career import Career
|
|
from schemas.career import CareerCreate, CareerUpdate
|
|
|
|
def get_all_careers(db: Session) -> List[Career]:
|
|
"""
|
|
Retrieves all careers from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
|
|
Returns:
|
|
List[Career]: A list of all career objects.
|
|
"""
|
|
return db.query(Career).all()
|
|
|
|
def get_career_by_id(db: Session, career_id: uuid.UUID) -> Optional[Career]:
|
|
"""
|
|
Retrieves a single career by its ID.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
career_id (UUID): The ID of the career to retrieve.
|
|
|
|
Returns:
|
|
Optional[Career]: The career object if found, otherwise None.
|
|
"""
|
|
return db.query(Career).filter(Career.id == career_id).first()
|
|
|
|
def get_careers_by_name(db: Session, name: str) -> List[Career]:
|
|
"""
|
|
Retrieves careers by name (case-insensitive search).
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
name (str): The name to search for.
|
|
|
|
Returns:
|
|
List[Career]: A list of career objects matching the name.
|
|
"""
|
|
return db.query(Career).filter(Career.name.ilike(f"%{name}%")).all()
|
|
|
|
def create_career(db: Session, career_data: CareerCreate) -> Career:
|
|
"""
|
|
Creates a new career in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
career_data (CareerCreate): The data for the career to create.
|
|
|
|
Returns:
|
|
Career: The newly created career object.
|
|
"""
|
|
db_career = Career(**career_data.dict())
|
|
db.add(db_career)
|
|
db.commit()
|
|
db.refresh(db_career)
|
|
return db_career
|
|
|
|
def update_career(db: Session, career_id: uuid.UUID, career_data: CareerUpdate) -> Optional[Career]:
|
|
"""
|
|
Updates an existing career in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
career_id (UUID): The ID of the career to update.
|
|
career_data (CareerUpdate): The updated data for the career.
|
|
|
|
Returns:
|
|
Optional[Career]: The updated career object if found, otherwise None.
|
|
"""
|
|
db_career = db.query(Career).filter(Career.id == career_id).first()
|
|
if not db_career:
|
|
return None
|
|
|
|
update_data = career_data.dict(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(db_career, key, value)
|
|
|
|
db.add(db_career)
|
|
db.commit()
|
|
db.refresh(db_career)
|
|
return db_career
|
|
|
|
def delete_career(db: Session, career_id: uuid.UUID) -> bool:
|
|
"""
|
|
Deletes a career from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
career_id (UUID): The ID of the career to delete.
|
|
|
|
Returns:
|
|
bool: True if the career was deleted, False otherwise.
|
|
"""
|
|
db_career = db.query(Career).filter(Career.id == career_id).first()
|
|
if not db_career:
|
|
return False
|
|
|
|
db.delete(db_career)
|
|
db.commit()
|
|
return True |