Spaces:
Running
Running
File size: 2,473 Bytes
bc1cd44 e1df50c bc1cd44 e1df50c bc1cd44 e1df50c bc1cd44 e1df50c bc1cd44 e1df50c bc1cd44 e1df50c bc1cd44 e1df50c bc1cd44 e1df50c bc1cd44 e1df50c bc1cd44 e1df50c bc1cd44 e1df50c bc1cd44 e1df50c bc1cd44 e1df50c bc1cd44 e1df50c bc1cd44 |
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 |
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 |