data/repository/annotator_repo.py

from typing import Optional
from sqlalchemy.orm import Session
import sentry_sdk

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}")
        #     sentry_sdk.capture_exception(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}")
        #     sentry_sdk.capture_exception(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}")
        #     sentry_sdk.capture_exception(exc)
        #     raise