data/repository/annotator_repo.py

from typing import Optional
from sqlalchemy.orm import Session

from data.models import Annotator
from utils.logger import Logger
from utils.security import hash_password

log = Logger()


class AnnotatorRepo:
    """
    Data-Access-Layer for Annotator table.
    """

    def __init__(self, db: Session) -> None:
        self.db = db

    # ------------------------------------------------------------------ #
    # READ METHODS
    # ------------------------------------------------------------------ #
    def get_annotator_by_name(self, name: str) -> Optional[Annotator]:
        try:
            return (
                self.db.query(Annotator)
                .filter(Annotator.name == name)
                .first()
            )
        except Exception as exc:
            log.error(f"Unable to fetch annotator <name={name}> : {exc}")
            raise

    def get_annotator_by_id(self, user_id: int) -> Optional[Annotator]:
        try:
            return (
                self.db.query(Annotator)
                .filter(Annotator.id == user_id)
                .first()
            )
        except Exception as exc:
            log.error(f"Unable to fetch annotator <id={user_id}> : {exc}")
            raise

    # ------------------------------------------------------------------ #
    # WRITE METHODS
    # ------------------------------------------------------------------ #
    def add_new_annotator(
        self,
        name: str,
        password: str,
        *,
        is_active: bool = True,
    ) -> Annotator:
        """
        Create a new Annotator with a hashed password.
        Raises:
            ValueError: if name already exists.
        """
        try:
            if self.get_annotator_by_name(name):
                raise ValueError(f"name `{name}` already exists.")

            # ------------------ HASH PASSWORD ------------------ #
            hashed_pass = hash_password(password)

            annotator = Annotator(
                name=name,
                password=hashed_pass,
                is_active=is_active,
            )
            self.db.add(annotator)
            self.db.flush()    # Ensure PK generated
            self.db.refresh(annotator)

            log.info(f"New annotator created <id={annotator.id} name={name}>")
            return annotator
        except Exception as exc:
            self.db.rollback()
            log.error(f"Unable to create annotator `{name}` : {exc}")
            raise