Spaces:
Running
Running
File size: 2,565 Bytes
bc1cd44 86cf81a bc1cd44 e1df50c 86cf81a bc1cd44 e1df50c 86cf81a bc1cd44 e1df50c bc1cd44 e1df50c bc1cd44 e1df50c bc1cd44 86cf81a bc1cd44 86cf81a bc1cd44 86cf81a bc1cd44 86cf81a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 |
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 |