27c9268a6a
- Set up project structure with FastAPI framework - Create database models for users, employees, departments, and job titles - Implement JWT authentication and authorization system - Set up SQLite database with SQLAlchemy ORM - Add Alembic migrations for database versioning - Create CRUD API endpoints for employee management - Implement category-based search functionality - Add OpenAPI documentation and health check endpoint - Update README with comprehensive setup and usage instructions
64 lines
2.8 KiB
Python
64 lines
2.8 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import or_
|
|
|
|
from app.crud.base import CRUDBase
|
|
from app.models.employee import Employee
|
|
from app.schemas.employee import EmployeeCreate, EmployeeUpdate, EmployeeSearchParams
|
|
|
|
|
|
class CRUDEmployee(CRUDBase[Employee, EmployeeCreate, EmployeeUpdate]):
|
|
def get_by_email(self, db: Session, *, email: str) -> Optional[Employee]:
|
|
return db.query(Employee).filter(Employee.email == email).first()
|
|
|
|
def get_all(self, db: Session, *, skip: int = 0, limit: int = 100) -> List[Employee]:
|
|
return db.query(Employee).offset(skip).limit(limit).all()
|
|
|
|
def get_by_department(self, db: Session, *, department_id: str, skip: int = 0, limit: int = 100) -> List[Employee]:
|
|
return db.query(Employee).filter(Employee.department_id == department_id).offset(skip).limit(limit).all()
|
|
|
|
def get_by_job_title(self, db: Session, *, job_title_id: str, skip: int = 0, limit: int = 100) -> List[Employee]:
|
|
return db.query(Employee).filter(Employee.job_title_id == job_title_id).offset(skip).limit(limit).all()
|
|
|
|
def get_by_manager(self, db: Session, *, manager_id: str, skip: int = 0, limit: int = 100) -> List[Employee]:
|
|
return db.query(Employee).filter(Employee.manager_id == manager_id).offset(skip).limit(limit).all()
|
|
|
|
def search(self, db: Session, *, params: EmployeeSearchParams, skip: int = 0, limit: int = 100) -> List[Employee]:
|
|
query = db.query(Employee)
|
|
|
|
# Apply filters based on the search parameters
|
|
if params.department_id:
|
|
query = query.filter(Employee.department_id == params.department_id)
|
|
|
|
if params.job_title_id:
|
|
query = query.filter(Employee.job_title_id == params.job_title_id)
|
|
|
|
if params.is_active is not None:
|
|
query = query.filter(Employee.is_active == params.is_active)
|
|
|
|
if params.name:
|
|
query = query.filter(
|
|
or_(
|
|
Employee.first_name.ilike(f"%{params.name}%"),
|
|
Employee.last_name.ilike(f"%{params.name}%")
|
|
)
|
|
)
|
|
|
|
if params.skills:
|
|
skill_filters = []
|
|
for skill in params.skills:
|
|
skill_filters.append(Employee.skills.ilike(f"%{skill}%"))
|
|
if skill_filters:
|
|
query = query.filter(or_(*skill_filters))
|
|
|
|
if params.categories:
|
|
category_filters = []
|
|
for category in params.categories:
|
|
category_filters.append(Employee.categories.ilike(f"%{category}%"))
|
|
if category_filters:
|
|
query = query.filter(or_(*category_filters))
|
|
|
|
return query.offset(skip).limit(limit).all()
|
|
|
|
|
|
employee = CRUDEmployee(Employee) |