molten_exp/market_look/manager.py

from abc import ABCMeta, abstractmethod
from inspect import Parameter
import typing

from sqlalchemy.orm import Session
from molten import BaseApp, HTTPError, HTTP_409, HTTP_404
from .errors import EntityNotFound
from .model import StoreModel, GeographyModel, AlertModel
from .schema import Alert, Geography, Link, Store, APIResponse


class BaseManager(metaclass=ABCMeta):
    """Base instance for Model managers"""

    def __init__(self, session: Session, app: BaseApp):
        self.session = session
        self.app = app

    @abstractmethod
    def _model_from_schema(self, schema):
        """Converts a Schema instance into a SQLAlchemy ORM model instance"""
        pass

    @abstractmethod
    def _schema_from_model(self, result):
        """Converts a SQLAlchemy results proxy into a Schema instance"""
        pass


class StoreManager(BaseManager):
    """A `StoreManager` is accountable for the CRUD operations associated with a Store"""

    def _schema_from_model(self, result: StoreModel) -> Store:
        _store = Store(
            id=result.id,
            href=self.app.reverse_uri("get_store_by_id", id=result.id),
            createdDate=result.created_date,
            modifiedDate=result.modified_date,
            name=result.name,
            number=result.number,
            address=result.address,
            city=result.city,
            state=result.state,
            zip=result.zip,
            lat=result.lat,
            long=result.long,
            tdlinx=result.tdlinx,
            geography=Link(
                href=self.app.reverse_uri("get_geo_by_id", id=result.geography_id)
            )
            if result.geography_id
            else None,
            alerts=Link(
                href=self.app.reverse_uri("get_store_alerts", id=result.id)
            )
        )
        return _store

    def _model_from_schema(self, store: Store) -> StoreModel:
        _store_model = StoreModel(
            id=store.id,
            name=store.name,
            number=store.number,
            address=store.address,
            city=store.city,
            state=store.state,
            zip=store.zip,
            lat=store.lat,
            long=store.long,
            tdlinx=store.tdlinx,
        )
        return _store_model

    def get_stores(self) -> typing.List[Store]:
        """Retrieves a list of Store representations"""
        results = self.session.query(StoreModel).order_by(StoreModel.id).all()
        _stores = [self._schema_from_model(result) for result in results]
        return _stores

    def get_store_by_id(self, id: int) -> Store:
        """Retrieves a store representation by id"""
        result = self.session.query(StoreModel).filter_by(id=id).one_or_none()
        if result is None:
            return result

        _store = self._schema_from_model(result)

        return _store

    def create_store(self, store: Store) -> Store:
        """Creates a new store resource and returns its representation"""
        result = self.session.query(StoreModel).filter_by(id=store.id).one_or_none()
        if result is not None:
            raise HTTPError(
                HTTP_409,
                {
                    "status": 409,
                    "message": f"A store with id: {store.id} already exists",
                },
            )

        store_model = self._model_from_schema(store)

        self.session.add(store_model)
        self.session.flush()

        _store = self._schema_from_model(store_model)

        return _store

    def update_store(self, store: Store) -> Store:
        result = self.session.query(StoreModel).filter_by(id=store.id).one_or_none()
        _updates = self._model_from_schema(store)
        self.session.merge(_updates)
        self.session.flush()
        _store = self._schema_from_model(result)
        return _store

    def delete_store_by_id(self, id):
        _store = self.session.query(StoreModel).filter_by(id=id).one_or_none()
        if _store is not None:
            self.session.delete(_store)
        return

    def get_stores_by_geo(self, geo_id) -> typing.List[Store]:
        results = self.session.query(StoreModel).filter_by(geography_id=geo_id).all()
        _stores = [self._schema_from_model(store) for store in results]
        return _stores


class GeographyManager(BaseManager):

    def _model_from_schema(self, schema: Geography) -> GeographyModel:
        pass

    def _schema_from_model(self, result: GeographyModel) -> Geography:
        _geography = Geography(
            id=result.id,
            href=self.app.reverse_uri("get_geo_by_id", id=result.id),
            createdDate=result.created_date,
            modifiedDate=result.modified_date,
            name=result.name,
            stores=Link(href=self.app.reverse_uri('get_geo_stores', id=result.id)),
            alerts=Link(href=self.app.reverse_uri('get_geo_alerts', id=result.id)),
        )
        return _geography

    def get_geographies(self) -> typing.List[Geography]:
        results = self.session.query(GeographyModel).all()
        return [self._schema_from_model(geo) for geo in results]

    def get_geo_by_id(self, id: int) -> Geography:
        result = self.session.query(GeographyModel).one_or_none()
        if result is None:
            raise HTTPError(
                HTTP_404,
                APIResponse(
                    status=404,
                    message=f"The resource you are looking for /geographies/{id} does not exist",
                ),
            )

        return self._schema_from_model(result)


class AlertManager(BaseManager):

    def _model_from_schema(self, schema: Alert) -> AlertModel:
        _alert_model = AlertModel(
            promo_name=schema.promoName,
            response=schema.response,
            valid=schema.valid
        )

    def _schema_from_model(self, result: AlertModel) -> Alert:
        _alert = Alert(
            id=result.id,
            href=self.app.reverse_uri('get_alert_by_id', id=result.id),
            createdDate=result.created_date,
            modifiedDate=result.modified_date,
            promoName=result.promo_name,
            response=result.response,
            valid=result.valid,
            store=Link(href=self.app.reverse_uri('get_store_by_id', id=result.store.id))
        )
        return _alert

    def get_alerts(self) -> typing.List[Alert]:
        results = self.session.query(AlertModel).all()
        alerts = [self._schema_from_model(alert) for alert in results]
        return alerts

    def get_alert_by_id(self, alert_id) -> Alert:
        result = self.session.query(AlertModel).filter_by(id=alert_id).one_or_none()
        if result is None:
            raise EntityNotFound(f"Alert: {alert_id} does not exist")
        alert = self._schema_from_model(result)
        return alert

    def get_alerts_at_store(self, store_id) -> typing.List[Alert]:
        store_check = self.session.query(StoreModel.id).filter_by(id=store_id).exists()

        if not self.session.query(store_check).scalar():
            raise EntityNotFound(f'Store {store_id} does not exist')

        results = self.session.query(AlertModel).filter_by(store_id=store_id, active=True).all()
        alerts = [self._schema_from_model(alert) for alert in results]
        return alerts

    def get_alerts_at_geo(self, geo_id) -> typing.List[Alert]:
        geo_check = self.session.query(GeographyModel.id).filter_by(id=geo_id).exists()

        if not self.session.query(geo_check).scalar():
            raise EntityNotFound(f'Geography {geo_id} does not exist')

        _subquery = self.session.query(StoreModel.id).filter(StoreModel.geography_id==geo_id)
        results = self.session.query(AlertModel).filter(AlertModel.store_id.in_(_subquery)).all()
        alerts = [self._schema_from_model(alert) for alert in results]
        return alerts

    def create_alert_at_store(self, store_id, alert: Alert) -> Alert:
        store = self.session.query(StoreModel).filter_by(id=store_id).one()
        alert_model = self._model_from_schema(alert)
        alert_model.store = store
        self.session.add(alert_model)
        self.session.flush()
        _alert = self._schema_from_model(alert_model)
        return _alert


class StoreManagerComponent:
    is_cacheable = True
    is_singleton = False

    def can_handle_parameter(self, parameter: Parameter) -> bool:
        return parameter.annotation is StoreManager

    def resolve(self, session: Session, app: BaseApp) -> StoreManager:  # type: ignore
        return StoreManager(session, app)


class GeographyManagerComponent:
    is_cacheable = True
    is_singleton = False

    def can_handle_parameter(self, parameter: Parameter) -> bool:
        return parameter.annotation is GeographyManager

    def resolve(self, session: Session, app: BaseApp) -> StoreManager:  # type: ignore
        return GeographyManager(session, app)


class AlertManagerComponent:
    is_cacheable = True
    is_singleton = False

    def can_handle_parameter(self, parameter: Parameter) -> bool:
        return parameter.annotation is AlertManager

    def resolve(self, session: Session, app: BaseApp) -> AlertManager:  # type: ignore
        return AlertManager(session, app)