molten_exp/market_look/app.py

import os
from typing import List, Tuple, Any, Dict, Optional
from molten import (
    annotate,
    App,
    Request,
    QueryParam,
    Route,
    Settings,
    SettingsComponent,
    ResponseRendererMiddleware,
    HTTP_201,
    HTTP_202,
    HTTP_404,
    HTTPError,
)
from molten.openapi import Metadata, OpenAPIHandler, OpenAPIUIHandler
from molten.contrib.sqlalchemy import (
    SQLAlchemyEngineComponent,
    SQLAlchemySessionComponent,
    SQLAlchemyMiddleware,
)

from .errors import EntityNotFound

from .manager import (
    AlertManager,
    StoreManager,
    GeographyManager,
    AlertManagerComponent,
    StoreManagerComponent,
    GeographyManagerComponent,
)
from .schema import Alert, Store, Geography, APIResponse
from .util import ExtJSONRender


class MarketLook(App):
    def handle_404(self, request: Request) -> Tuple[str, APIResponse]:
        return (
            HTTP_404,
            APIResponse(
                status=404,
                message=f"The resource you are looking for {request.scheme}://{request.host}{request.path} doesn't exist",
            ),
        )


# We are passing explicitly the engine param for
# establishing the utc as the timezone for our connection.
settings = Settings(
    {
        "database_engine_dsn": os.getenv(
            "DATABASE_DSN", "postgres://molten:local@localhost/market_look"
        ),
        "database_engine_params": {
            "echo": True,
            "connect_args": {"options": "-c timezone=utc"},
        },
    }
)

get_schema = OpenAPIHandler(
    metadata=Metadata(
        title="Market Look API",
        description="An API for managing promotion compliance at store.",
        version="0.1.0",
    )
)

get_docs = OpenAPIUIHandler()
setattr(get_docs, "openapi_tags", ["API Management"])


def name(name: Optional[QueryParam]) -> APIResponse:
    _name = name or "Molten"
    return APIResponse(
        status=200, message=f"Hello, {_name}! Glad you are programming with us"
    )


def ping() -> APIResponse:
    return APIResponse(200, "Pong")


@annotate(openapi_tags=["Stores"])
def get_stores(store_manager: StoreManager) -> List[Store]:
    """Returns a collection of Stores"""
    stores = store_manager.get_stores()
    return stores


@annotate(openapi_tags=["Stores"])
def create_store(
        store: Store, store_manager: StoreManager
) -> Tuple[Any, Store, Dict[str, str]]:
    """Creates a new store resource and returns its respresentation"""
    try:
        _store = store_manager.create_store(store)
    except HTTPError as err:
        raise err
    headers = {"Location": _store.href}
    return HTTP_201, _store, headers


@annotate(openapi_tags=["Stores"])
def get_store_by_id(id: int, store_manager: StoreManager) -> Store:
    _store = store_manager.get_store_by_id(id)

    if _store is None:
        raise HTTPError(
            HTTP_404,
            {
                "status": 404,
                "message": f"The resource you are looking for /stores/{id} does not exist",
            },
        )

    return _store


@annotate(openapi_tags=["Stores"])
def update_store(store: Store, store_manager: StoreManager) -> Store:
    _updated_store = store_manager.update_store(store)
    return _updated_store


@annotate(openapi_tags=["Stores"])
def delete_store_by_id(id: int, store_manager: StoreManager) -> Tuple[Any, APIResponse]:
    store_manager.delete_store_by_id(id)
    return (
        HTTP_202,
        APIResponse(status=202, message=f"Delete request for store: {id} accepted"),
    )


@annotate(openapi_tags=["Geographies"])
def get_geographies(geo_manager: GeographyManager) -> List[Geography]:
    _geographies = geo_manager.get_geographies()
    return _geographies


@annotate(openapi_tags=["Geographies"])
def get_geo_by_id(id: int, geo_manager: GeographyManager) -> Geography:
    try:
        _geo = geo_manager.get_geo_by_id(id=id)
    except HTTPError as err:
        raise err

    return _geo


@annotate(openapi_tags=["Geographies"])
def get_geo_stores(id: int, store_manager: StoreManager) -> List[Store]:
    stores = store_manager.get_stores_by_geo(geo_id=id)
    return stores


@annotate(openapi_tags=["Alerts"])
def get_alerts(alert_manager: AlertManager) -> List[Alert]:
    alerts = alert_manager.get_alerts()
    return alerts


@annotate(openapi_tags=["Alerts"])
def get_alert_by_id(id: int, alert_manager: AlertManager) -> Alert:
    try:
        alert = alert_manager.get_alert_by_id(alert_id=id)
    except EntityNotFound as err:
        raise HTTPError(HTTP_404, APIResponse(status=404, message=err.message))
    return alert


@annotate(openapi_tags=["Alerts"])
def get_store_alerts(id: int, alert_manager: AlertManager) -> List[Alert]:
    try:
        alerts = alert_manager.get_alerts_at_store(store_id=id)
    except EntityNotFound as err:
        raise HTTPError(HTTP_404, APIResponse(status=404, message=err.message))
    return alerts


@annotate(openapi_tags=["Alerts"])
def get_geo_alerts(id: int, alert_manager: AlertManager) -> List[Alert]:
    try:
        alerts = alert_manager.get_alerts_at_geo(geo_id=id)
    except EntityNotFound as err:
        raise HTTPError(HTTP_404, APIResponse(status=404, message=err.message))
    return alerts


routes = [
    Route("/", method="GET", handler=name),
    Route("/ping", method="GET", handler=ping),
    # Stores
    Route("/stores", method="GET", handler=get_stores),
    Route("/stores", method="POST", handler=create_store),
    Route("/stores/{id}", method="GET", handler=get_store_by_id),
    Route("/stores/{id}", method="PUT", handler=update_store),
    Route("/stores/{id}", method="DELETE", handler=delete_store_by_id),
    Route("/geographies/{id}/stores", method="GET", handler=get_geo_stores),
    # Geographies
    Route("/geographies", method="GET", handler=get_geographies),
    Route("/geographies/{id}", method="GET", handler=get_geo_by_id),
    # Alerts
    Route("/alerts", method="GET", handler=get_alerts),
    Route("/alerts/{id}", method="GET", handler=get_alert_by_id),
    Route("/stores/{id}/alerts", method="GET", handler=get_store_alerts),
    Route("/geographies/{id}/alerts", method="GET", handler=get_geo_alerts),
    # OpenAPI
    Route("/_schema", get_schema),
    Route("/docs", get_docs),
]

components = [
    SettingsComponent(settings),
    SQLAlchemyEngineComponent(),
    SQLAlchemySessionComponent(),
    StoreManagerComponent(),
    GeographyManagerComponent(),
    AlertManagerComponent()
]

middleware = [ResponseRendererMiddleware(), SQLAlchemyMiddleware()]

renderers = [ExtJSONRender()]

app = MarketLook(
    routes=routes, components=components, middleware=middleware, renderers=renderers
)