cookie-api/cookie_api/app.py

from apistar import Include, Route, http, annotate
from apistar.backends import sqlalchemy_backend
from apistar.backends.sqlalchemy_backend import Session
from apistar.frameworks.wsgi import WSGIApp as App
from apistar.handlers import docs_urls, static_urls
from apistar.interfaces import Router, Injector, Auth
from apistar_jwt.authentication import JWTAuthentication

from cookie_api.auth import auth_routes, auth_components
from cookie_api.commands import commands
from cookie_api.models import Cookie
from cookie_api.schema import CookieSchema

cookie_schema = CookieSchema()


@annotate(authentication=[JWTAuthentication()])
def get_state(injector: Injector, auth: Auth):
    state = injector.state
    d = dict()
    for k, v in state.items():
        d[k] = str(v)
    return d


def get_cookies(session: Session):
    """Retrieves all Cookies"""
    cookies = session.query(Cookie).all()
    return cookie_schema.dump(cookies, many=True).data


def create_cookie(session: Session, json_data: http.RequestData, route: Router):
    """Create a Cookie"""
    cookie_data, errors = cookie_schema.load(json_data)
    if errors:
        msg = {"message": "400 Bad Request", "error": errors}
        return http.Response(msg, status=400)
    cookie = Cookie(**cookie_data)
    session.add(cookie)
    session.flush()
    headers = {'Location': route.reverse_url('get_cookie', dict(id=cookie.id))}
    return http.Response(cookie_schema.dump(cookie).data, status=201, headers=headers)


def get_cookie(session: Session, id):
    """Retrieve Cookie by id"""
    cookie = session.query(Cookie).filter_by(id=id).one_or_none()
    if cookie is None:
        msg = {"message": "404 Not Found"}
        return http.Response(msg, status=404)
    return cookie_schema.dump(cookie).data


def patch_cookie(session: Session, json_data: http.RequestData, id: int):
    """Update Cookie through partial update"""
    cookie = session.query(Cookie).filter_by(id=id).one_or_none()
    if cookie is None:
        msg = {"message": "404 Not Found"}
        return http.Response(msg, status=404)
    # run schema validation of fields
    cookie_data, errors = cookie_schema.load(json_data, partial=True)
    if errors:
        msg = {"message": "400 Bad Request", "error": errors}
        return http.Response(msg, status=400)

    for k, v in cookie_data.items():
        setattr(cookie, k, v)
    session.flush()
    return cookie_schema.dump(cookie).data


def delete_cookie(session: Session, id: int):
    """Delete a Cookie"""
    cookie = session.query(Cookie).filter_by(id=id).one_or_none()
    if cookie is None:
        msg = {"message": "404 Not Found"}
        return http.Response(msg, status=404)
    session.delete(cookie)
    return {"message": "200 OK"}


routes = [
    Route('/state', 'GET', get_state),
    Route('/cookies', 'GET', get_cookies),
    Route('/cookies', 'POST', create_cookie),
    Route('/cookies/{id}', 'GET', get_cookie),
    Route('/cookies/{id}', 'PATCH', patch_cookie),
    Route('/cookies/{id}', 'DELETE', delete_cookie),
    Include('/docs', docs_urls),
    Include('/static', static_urls)
]

app_settings = {}

routes = routes + auth_routes

commands = sqlalchemy_backend.commands + commands

components = sqlalchemy_backend.components + auth_components


def application_factory(settings={}):
    """Returns an instance of Cookie API"""
    return App(settings={**app_settings, **settings},
               commands=commands,
               components=components,
               routes=routes)