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)