romm/backend/handler/database/container_adoptions_handler.py

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from decorators.database import INJECTED_SESSION, begin_session
from models.container_adoption import AdoptionOutcome, StreamingContainerAdoption

from .base_handler import DBBaseHandler


class DBContainerAdoptionsHandler(DBBaseHandler):
    @begin_session
    def get_adoption(
        self,
        container_key: str,
        session: Session = INJECTED_SESSION,
    ) -> StreamingContainerAdoption | None:
        return session.scalar(
            select(StreamingContainerAdoption).filter_by(container_key=container_key)
        )

    @begin_session
    def add_adoption(
        self,
        container_key: str,
        outcome: AdoptionOutcome,
        user_id: int,
        session: Session = INJECTED_SESSION,
    ) -> StreamingContainerAdoption | None:
        """Returns None when another claim recorded the decision first. The
        unique constraint is the arbiter, not the earlier read."""
        adoption = StreamingContainerAdoption(
            container_key=container_key,
            outcome=outcome,
            decided_by_user_id=user_id,
        )
        try:
            session.add(adoption)
            session.flush()
        except IntegrityError:
            session.rollback()
            return None
        return adoption