61 lines
1.9 KiB
Python
61 lines
1.9 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Repository für Employee-CRUD – einziger Ort der direkt mit der DB spricht."""
|
||
|
||
from typing import Optional
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..core.models import Employee
|
||
from ..core.schemas import EmployeeCreate, EmployeeUpdate
|
||
|
||
|
||
class EmployeeRepository:
|
||
def __init__(self, db: Session):
|
||
self.db = db
|
||
|
||
def get_by_id(self, employee_id: str) -> Optional[Employee]:
|
||
return self.db.query(Employee).filter(Employee.id == employee_id).first()
|
||
|
||
def get_by_email(self, email: str) -> Optional[Employee]:
|
||
return self.db.query(Employee).filter(Employee.email == email).first()
|
||
|
||
def list_all(self, active_only: bool = True) -> list[Employee]:
|
||
q = self.db.query(Employee)
|
||
if active_only:
|
||
q = q.filter(Employee.is_active == True)
|
||
return q.order_by(Employee.name).all()
|
||
|
||
def list_by_department(self, department: str) -> list[Employee]:
|
||
return (
|
||
self.db.query(Employee)
|
||
.filter(Employee.department == department, Employee.is_active == True)
|
||
.order_by(Employee.name)
|
||
.all()
|
||
)
|
||
|
||
def create(self, data: EmployeeCreate) -> Employee:
|
||
emp = Employee(**data.model_dump())
|
||
self.db.add(emp)
|
||
self.db.flush()
|
||
return emp
|
||
|
||
def update(self, employee_id: str, data: EmployeeUpdate) -> Optional[Employee]:
|
||
emp = self.get_by_id(employee_id)
|
||
if not emp:
|
||
return None
|
||
for field, value in data.model_dump(exclude_unset=True).items():
|
||
setattr(emp, field, value)
|
||
self.db.flush()
|
||
return emp
|
||
|
||
def delete(self, employee_id: str) -> bool:
|
||
emp = self.get_by_id(employee_id)
|
||
if not emp:
|
||
return False
|
||
self.db.delete(emp)
|
||
self.db.flush()
|
||
return True
|
||
|
||
def count_active(self) -> int:
|
||
return self.db.query(Employee).filter(Employee.is_active == True).count()
|