feat: add login with username & password

pyproject.toml

@@ -26,7 +26,9 @@ robonomics-interface = "^1.2.2"
 ecs-logging = "^2.0.0"
 aioprometheus = "^22.5.0"
 pycups = "^2.0.1"
-pydantic-settings = "^2.2.1"
+passlib = {extras = ["bcrypt"], version = "^1.7.4"}
+pydantic_settings = "^2.2.1"
+
 
 [tool.poetry.dev-dependencies]
 mypy = "^0.971"

src/database/models.py

@@ -41,6 +41,7 @@ class EmployeeModel(BaseModel):
 
 class EmployeeWCardModel(EmployeeModel):
     rfid_card_id: str | None
+    username: str | None
 
 
 class AdditionalDetail:
@@ -70,6 +71,11 @@ class EmployeeID(BaseModel):
     employee_rfid_card_no: str
 
 
+class EmployeeCreds(BaseModel):
+    employee_username: str
+    employee_password: str
+
+
 class UnitOut(GenericResponse):
     unit_internal_id: str | None
 

src/dependencies.py

@@ -36,6 +36,18 @@ def get_employee_by_card_id(employee_data: models.EmployeeID) -> models.Employee
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) from e
 
 
+def get_employee_by_username(employee_data: models.EmployeeCreds) -> models.EmployeeWCardModel:
+    try:
+        employee: Employee = employee_wrapper.get_employee_by_username(
+            username=employee_data.employee_username, password=employee_data.employee_password
+        )
+        return models.EmployeeWCardModel(**asdict(employee))
+
+    except EmployeeNotFoundError as e:
+        messenger.warning(translation("NoEmployee"))
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) from e
+
+
 def get_schema_by_id(schema_id: str) -> models.ProductionSchema:
     """get the specified production schema"""
     try:

src/employee/Employee.py

@@ -2,14 +2,18 @@
 from dataclasses import dataclass
 
 
+
 @dataclass
 class Employee:
-    rfid_card_id: str
     name: str
     position: str
+    rfid_card_id: str | None = None
+    username: str | None = None
     passport_code: str = ""
 
     def __post_init__(self) -> None:
+        if not self.username and not self.rfid_card_id:
+            raise ValueError("needs username or card id")
         if not self.passport_code:
             self.passport_code = self.get_passport_code()
 

src/employee/employee_wrapper.py

@@ -5,6 +5,9 @@
 from src.feecc_workbench.exceptions import EmployeeNotFoundError
 from src.feecc_workbench.Types import Document
 from .Employee import Employee
+from passlib.context import CryptContext
+
+pwd_context = CryptContext(schemes=["bcrypt"])
 
 
 class EmployeeWrapper:
@@ -14,15 +17,39 @@ class EmployeeWrapper:
     def get_employee_by_card_id(self, card_id: str) -> Employee:
         """find the employee with the provided RFID card id"""
         filters = {"rfid_card_id": card_id}
-        projection = {"_id": 0}
-        employee_data: list[Document] | None = BaseMongoDbWrapper.find(collection=self.collection, filters=filters, projection=projection)
+        projection = {"_id": 0, "hashed_password": 0}
+        employee_data = BaseMongoDbWrapper.find_one(collection=self.collection, filters=filters, projection=projection)
 
         if employee_data is None:
             message = f"No employee with card ID {card_id}"
             logger.error(message)
             raise EmployeeNotFoundError(message)
 
-        return Employee(**employee_data[0])
+        return Employee(**employee_data)
+
+    @time_execution
+    def get_employee_by_username(self, username: str, password: str) -> Employee:
+        """find the employee with the provided RFID card id"""
+        filters = {"username": username}
+        projection = {"_id": 0}
+        employee_data = BaseMongoDbWrapper.find_one(collection=self.collection, filters=filters, projection=projection)
+
+        if employee_data is None:
+            message = f"No employee with username {username}"
+            logger.error(message)
+            raise EmployeeNotFoundError(message)
+
+        if not bool(pwd_context.verify(password, employee_data["hashed_password"])):
+            message = f"Incorrect password for username {username}"
+            logger.error(message)
+            raise EmployeeNotFoundError(message)
+
+        return Employee(
+            rfid_card_id=employee_data["rfid_card_id"],
+            name=employee_data["name"],
+            position=employee_data["position"],
+            username=employee_data["username"],
+        )
 
 
 employee_wrapper = EmployeeWrapper()

src/routers/employee_router.py

@@ -2,7 +2,7 @@
 from loguru import logger
 from starlette import status
 
-from src.dependencies import get_employee_by_card_id
+from src.dependencies import get_employee_by_card_id, get_employee_by_username
 from src.database import models as mdl
 from src.employee.Employee import Employee
 from src.feecc_workbench.exceptions import StateForbiddenError
@@ -26,13 +26,39 @@ def get_employee_data(
     )
 
 
+@router.post("/login-creds", response_model=mdl.EmployeeOut)
+def log_in_creds(employee: mdl.EmployeeWCardModel = Depends(get_employee_by_username)) -> mdl.EmployeeOut:
+    try:
+        WORKBENCH.log_in(
+            Employee(
+                rfid_card_id=employee.rfid_card_id,
+                name=employee.name,
+                position=employee.position,
+                username=employee.username,
+            )
+        )
+        return mdl.EmployeeOut(
+            status_code=status.HTTP_200_OK, detail="Employee logged in successfully", employee_data=employee
+        )
+
+    except StateForbiddenError as e:
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(e)) from e
+
+
 @router.post("/log-in", response_model=mdl.EmployeeOut)
 def log_in_employee(
     employee: mdl.EmployeeWCardModel = Depends(get_employee_by_card_id),  # noqa: B008
 ) -> mdl.EmployeeOut:
     """handle logging in the Employee at a given Workbench"""
     try:
-        WORKBENCH.log_in(Employee(rfid_card_id=employee.rfid_card_id, name=employee.name, position=employee.position))
+        WORKBENCH.log_in(
+            Employee(
+                rfid_card_id=employee.rfid_card_id,
+                name=employee.name,
+                position=employee.position,
+                username=employee.username,
+            )
+        )
         return mdl.EmployeeOut(
             status_code=status.HTTP_200_OK, detail="Employee logged in successfully", employee_data=employee
         )