114 lines
3.5 KiB
Python
114 lines
3.5 KiB
Python
from typing import List, Optional
|
|
import uuid
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import or_
|
|
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.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_or_description(db: Session, search_term: str, tech_career: Optional[bool] = None) -> List[Career]:
|
|
"""
|
|
Retrieves careers that match the given search term in their name or description, optionally filtering by tech career.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
search_term (str): The search term to match against career names and descriptions.
|
|
tech_career (Optional[bool]): Whether to filter for tech careers or non-tech careers. If None, no filtering is applied.
|
|
|
|
Returns:
|
|
List[Career]: A list of career objects that match the search term and tech career filter.
|
|
"""
|
|
query = db.query(Career).filter(
|
|
or_(
|
|
Career.name.ilike(f"%{search_term}%"),
|
|
Career.description.ilike(f"%{search_term}%")
|
|
)
|
|
)
|
|
if tech_career is not None:
|
|
query = query.filter(Career.tech_career == tech_career)
|
|
return query.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.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 field, value in update_data.items():
|
|
setattr(db_career, field, 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.UUID): The ID of the career to delete.
|
|
|
|
Returns:
|
|
bool: True if the career was successfully 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 |