flaschengeist-schedule/backend/flaschengeist_events/routes.py

529 lines
16 KiB
Python

from http.client import NO_CONTENT
from flask import request, jsonify
from werkzeug.exceptions import BadRequest, NotFound, Forbidden
from flaschengeist.models.session import Session
from flaschengeist.controller import userController
from flaschengeist.utils.decorators import login_required
from flaschengeist.utils.datetime import from_iso_format
from flaschengeist.utils.HTTP import get_filter_args, no_content
from . import event_controller, permissions, EventPlugin
def dict_get(self, key, default=None, type=None):
"""Same as .get from MultiDict"""
try:
rv = self[key]
except KeyError:
return default
if type is not None:
try:
rv = type(rv)
except ValueError:
rv = default
return rv
@EventPlugin.blueprint.route("/events/templates", methods=["GET"])
@login_required()
def get_templates(current_session):
return jsonify(event_controller.get_templates())
@EventPlugin.blueprint.route("/events/event-types", methods=["GET"])
@EventPlugin.blueprint.route("/events/event-types/<int:identifier>", methods=["GET"])
@login_required()
def get_event_types(current_session, identifier=None):
"""Get EventType(s)
Route: ``/events/event-types`` | Method: ``GET``
Route: ``/events/event-types/<identifier>`` | Method: ``GET``
Args:
current_session: Session sent with Authorization Header
identifier: If querying a specific EventType
Returns:
JSON encoded (list of) EventType(s) or HTTP-error
"""
if identifier:
result = event_controller.get_event_type(identifier)
else:
result = event_controller.get_event_types()
return jsonify(result)
@EventPlugin.blueprint.route("/events/event-types", methods=["POST"])
@login_required(permission=permissions.EVENT_TYPE)
def new_event_type(current_session):
"""Create a new EventType
Route: ``/events/event-types`` | Method: ``POST``
POST-data: ``{name: string}``
Args:
current_session: Session sent with Authorization Header
Returns:
HTTP-Created or HTTP-error
"""
data = request.get_json()
if "name" not in data:
raise BadRequest
event_type = event_controller.create_event_type(data["name"])
return jsonify(event_type)
@EventPlugin.blueprint.route("/events/event-types/<int:identifier>", methods=["PUT", "DELETE"])
@login_required(permission=permissions.EVENT_TYPE)
def modify_event_type(identifier, current_session):
"""Rename or delete an event type
Route: ``/events/event-types/<id>`` | Method: ``PUT`` or ``DELETE``
POST-data: (if renaming) ``{name: string}``
Args:
identifier: Identifier of the EventType
current_session: Session sent with Authorization Header
Returns:
HTTP-NoContent or HTTP-error
"""
if request.method == "DELETE":
event_controller.delete_event_type(identifier)
else:
data = request.get_json()
if "name" not in data:
raise BadRequest("Parameter missing in data")
event_controller.rename_event_type(identifier, data["name"])
return "", NO_CONTENT
@EventPlugin.blueprint.route("/events/job-types", methods=["GET"])
@login_required()
def get_job_types(current_session):
"""Get all JobTypes
Route: ``/events/job-types`` | Method: ``GET``
Args:
current_session: Session sent with Authorization Header
Returns:
JSON encoded list of JobType HTTP-error
"""
types = event_controller.get_job_types()
return jsonify(types)
@EventPlugin.blueprint.route("/events/job-types", methods=["POST"])
@login_required(permission=permissions.JOB_TYPE)
def new_job_type(current_session):
"""Create a new JobType
Route: ``/events/job-types`` | Method: ``POST``
POST-data: ``{name: string}``
Args:
current_session: Session sent with Authorization Header
Returns:
JSON encoded JobType or HTTP-error
"""
data = request.get_json()
if "name" not in data:
raise BadRequest
jt = event_controller.create_job_type(data["name"])
return jsonify(jt)
@EventPlugin.blueprint.route("/events/job-types/<int:type_id>", methods=["PUT", "DELETE"])
@login_required(permission=permissions.JOB_TYPE)
def modify_job_type(type_id, current_session):
"""Rename or delete a JobType
Route: ``/events/job-types/<name>`` | Method: ``PUT`` or ``DELETE``
POST-data: (if renaming) ``{name: string}``
Args:
type_id: Identifier of the JobType
current_session: Session sent with Authorization Header
Returns:
HTTP-NoContent or HTTP-error
"""
if request.method == "DELETE":
event_controller.delete_job_type(type_id)
else:
data = request.get_json()
if "name" not in data:
raise BadRequest("Parameter missing in data")
event_controller.rename_job_type(type_id, data["name"])
return "", NO_CONTENT
@EventPlugin.blueprint.route("/events/<int:event_id>", methods=["GET"])
@login_required()
def get_event(event_id, current_session):
"""Get event by id
Route: ``/events/<event_id>`` | Method: ``GET``
Args:
event_id: ID identifying the event
current_session: Session sent with Authorization Header
Returns:
JSON encoded event object
"""
event = event_controller.get_event(
event_id,
with_backup=current_session.user_.has_permission(permissions.SEE_BACKUP),
)
return jsonify(event)
@EventPlugin.blueprint.route("/events", methods=["GET"])
@login_required()
def get_events(current_session):
count, result = event_controller.get_events(
*get_filter_args(),
with_backup=current_session.user_.has_permission(permissions.SEE_BACKUP),
)
return jsonify({"count": count, "result": result})
def _add_job(event, data):
try:
start = from_iso_format(data["start"])
end = dict_get(data, "end", None, type=from_iso_format)
required_services = data["required_services"]
job_type = data["type"]
if isinstance(job_type, dict):
job_type = data["type"]["id"]
except (KeyError, ValueError):
raise BadRequest("Missing or invalid POST parameter")
job_type = event_controller.get_job_type(job_type)
event_controller.add_job(
event,
job_type,
required_services,
start,
end,
comment=dict_get(data, "comment", None, str),
)
@EventPlugin.blueprint.route("/events", methods=["POST"])
@login_required(permission=permissions.CREATE)
def create_event(current_session):
"""Create an new event
Route: ``/events`` | Method: ``POST``
POST-data: See interfaces for Event, can already contain jobs
Args:
current_session: Session sent with Authorization Header
Returns:
JSON encoded Event object or HTTP-error
"""
data = request.get_json()
try:
start = from_iso_format(data["start"])
end = dict_get(data, "end", None, type=from_iso_format)
data_type = data["type"]
if isinstance(data_type, dict):
data_type = data["type"]["id"]
event_type = event_controller.get_event_type(data_type)
except KeyError:
raise BadRequest("Missing POST parameter")
except (NotFound, ValueError):
raise BadRequest("Invalid parameter")
event = event_controller.create_event(
start=start,
end=end,
name=dict_get(data, "name", None),
is_template=dict_get(data, "is_template", None),
event_type=event_type,
description=dict_get(data, "description", None),
)
if "jobs" in data:
for job in data["jobs"]:
_add_job(event, job)
return jsonify(event)
@EventPlugin.blueprint.route("/events/<int:event_id>", methods=["PUT"])
@login_required(permission=permissions.EDIT)
def modify_event(event_id, current_session):
"""Modify an event
Route: ``/events/<event_id>`` | Method: ``PUT``
POST-data: See interfaces for Event, can already contain slots
Args:
event_id: Identifier of the event
current_session: Session sent with Authorization Header
Returns:
JSON encoded Event object or HTTP-error
"""
event = event_controller.get_event(event_id)
data = request.get_json()
event.start = dict_get(data, "start", event.start, type=from_iso_format)
event.end = dict_get(data, "end", event.end, type=from_iso_format)
event.name = dict_get(data, "name", event.name, type=str)
event.description = dict_get(data, "description", event.description, type=str)
if "type" in data:
event_type = event_controller.get_event_type(data["type"])
event.type = event_type
event_controller.update()
return jsonify(event)
@EventPlugin.blueprint.route("/events/<int:event_id>", methods=["DELETE"])
@login_required(permission=permissions.DELETE)
def delete_event(event_id, current_session):
"""Delete an event
Route: ``/events/<event_id>`` | Method: ``DELETE``
Args:
event_id: Identifier of the event
current_session: Session sent with Authorization Header
Returns:
HTTP-NoContent or HTTP-error
"""
event_controller.delete_event(event_id)
return "", NO_CONTENT
@EventPlugin.blueprint.route("/events/<int:event_id>/jobs", methods=["POST"])
@login_required(permission=permissions.EDIT)
def add_job(event_id, current_session):
"""Add an new Job to an Event / EventSlot
Route: ``/events/<event_id>/jobs`` | Method: ``POST``
POST-data: See Job
Args:
event_id: Identifier of the event
current_session: Session sent with Authorization Header
Returns:
JSON encoded Event object or HTTP-error
"""
event = event_controller.get_event(event_id)
_add_job(event, request.get_json())
return jsonify(event)
@EventPlugin.blueprint.route("/events/<int:event_id>/jobs/<int:job_id>", methods=["DELETE"])
@login_required(permission=permissions.DELETE)
def delete_job(event_id, job_id, current_session):
"""Delete a Job
Route: ``/events/<event_id>/jobs/<job_id>`` | Method: ``DELETE``
Args:
event_id: Identifier of the event
job_id: Identifier of the Job
current_session: Session sent with Authorization Header
Returns:
HTTP-no-content or HTTP error
"""
job = event_controller.get_job(job_id, event_id)
event_controller.delete_job(job)
return no_content()
@EventPlugin.blueprint.route("/events/<int:event_id>/jobs/<int:job_id>", methods=["PUT"])
@login_required()
def update_job(event_id, job_id, current_session: Session):
"""Edit Job
Route: ``/events/<event_id>/jobs/<job_id>`` | Method: ``PUT``
POST-data: See TS interface for Job
Args:
event_id: Identifier of the event
job_id: Identifier of the Job
current_session: Session sent with Authorization Header
Returns:
JSON encoded Job object or HTTP-error
"""
if not current_session.user_.has_permission(permissions.EDIT):
raise Forbidden
data = request.get_json()
if not data:
raise BadRequest
job = event_controller.get_job(job_id, event_id)
try:
if "type" in data:
job.type = event_controller.get_job_type(data["type"])
job.start = from_iso_format(data.get("start", job.start))
job.end = from_iso_format(data.get("end", job.end))
job.comment = str(data.get("comment", job.comment))
job.locked = bool(data.get("locked", job.locked))
job.required_services = float(data.get("required_services", job.required_services))
event_controller.update()
except NotFound:
raise BadRequest("Invalid JobType")
except ValueError:
raise BadRequest("Invalid POST data")
return jsonify(job)
@EventPlugin.blueprint.route("/events/jobs", methods=["GET"])
@login_required()
def get_jobs(current_session: Session):
count, result = event_controller.get_jobs(current_session.user_, *get_filter_args())
return jsonify({"count": count, "result": result})
@EventPlugin.blueprint.route("/events/jobs/<int:job_id>/assign", methods=["POST"])
@login_required()
def assign_job(job_id, current_session: Session):
"""Assign / unassign user to the Job
Route: ``/events/jobs/<job_id>/assign`` | Method: ``POST``
POST-data: a Service object, see TS interface for Service
Args:
job_id: Identifier of the Job
current_session: Session sent with Authorization Header
Returns:
JSON encoded Job or HTTP-error
"""
data = request.get_json()
job = event_controller.get_job(job_id)
try:
user = userController.get_user(data["userid"])
value = data["value"]
if (user == current_session.user_ and not user.has_permission(permissions.ASSIGN)) or (
user != current_session.user_ and not current_session.user_.has_permission(permissions.ASSIGN_OTHER)
):
raise Forbidden
if value > 0:
event_controller.assign_job(job, user, value, data.get("is_backup", False))
else:
event_controller.unassign_job(job, user, notify=user != current_session.user_)
except (TypeError, KeyError, ValueError):
raise BadRequest
return jsonify(job)
@EventPlugin.blueprint.route("/events/jobs/<int:job_id>/lock", methods=["POST"])
@login_required(permissions.LOCK_JOBS)
def lock_job(job_id, current_session: Session):
"""Lock / unlock the Job
Route: ``/events/jobs/<job_id>/lock`` | Method: ``POST``
POST-data: ``{locked: boolean}``
Args:
job_id: Identifier of the Job
current_session: Session sent with Authorization Header
Returns:
HTTP-No-Content or HTTP-error
"""
data = request.get_json()
job = event_controller.get_job(job_id)
try:
locked = bool(userController.get_user(data["locked"]))
job.locked = locked
event_controller.update()
except (TypeError, KeyError, ValueError):
raise BadRequest
return no_content()
@EventPlugin.blueprint.route("/events/invitations", methods=["POST"])
@login_required()
def invite(current_session: Session):
"""Invite an user to a job or transfer job
Route: ``/events/invites`` | Method: ``POST``
POST-data: ``{job: number, invitees: string[], is_transfer?: boolean}``
Args:
current_session: Session sent with Authorization Header
Returns:
List of Invitation objects or HTTP-error
"""
data = request.get_json()
transferee = data.get("transferee", None)
if (
transferee is not None
and transferee != current_session.userid
and not current_session.user_.has_permission(permissions.ASSIGN_OTHER)
):
raise Forbidden
try:
job = event_controller.get_job(data["job"])
if not isinstance(data["invitees"], list):
raise BadRequest
return jsonify(
[
event_controller.invite(job, invitee, current_session.user_, transferee)
for invitee in [userController.get_user(uid) for uid in data["invitees"]]
]
)
except (TypeError, KeyError, ValueError):
raise BadRequest
@EventPlugin.blueprint.route("/events/invitations/<int:invitation_id>", methods=["GET"])
@login_required()
def get_invitation(invitation_id: int, current_session: Session):
inv = event_controller.get_invitation(invitation_id)
if current_session.userid not in [inv.invitee_id, inv.inviter_id, inv.transferee_id]:
raise Forbidden
return jsonify(inv)
@EventPlugin.blueprint.route("/events/invitations/<int:invitation_id>", methods=["DELETE", "PUT"])
@login_required()
def respond_invitation(invitation_id: int, current_session: Session):
inv = event_controller.get_invitation(invitation_id)
if request.method == "DELETE":
if current_session.userid == inv.invitee_id:
event_controller.respond_invitation(inv, False)
elif current_session.userid == inv.inviter_id:
event_controller.cancel_invitation(inv)
else:
raise Forbidden
else:
# maybe validate data is something like ({accepted: true})
if current_session.userid != inv.invitee_id:
raise Forbidden
event_controller.respond_invitation(inv)
return no_content()