You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
108 lines
3.4 KiB
Python
108 lines
3.4 KiB
Python
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)
|