from http.client import NO_CONTENT from flask import request, jsonify, Blueprint 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 blueprint = Blueprint("events", __name__) 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 @blueprint.route("/events/templates", methods=["GET"]) @login_required() def get_templates(current_session): return jsonify(event_controller.get_templates()) @blueprint.route("/events/event-types", methods=["GET"]) @blueprint.route("/events/event-types/", 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/`` | 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) @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) @blueprint.route("/events/event-types/", 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/`` | 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 @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) @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) @blueprint.route("/events/job-types/", 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/`` | 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 @blueprint.route("/events/", methods=["GET"]) @login_required() def get_event(event_id, current_session): """Get event by id Route: ``/events/`` | 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) @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 = int(data["type"]) 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), ) @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) event_type = event_controller.get_event_type(int(data["type"])) event = event_controller.create_event( start=start, end=end, name=dict_get(data, "name", None, type=str), is_template=dict_get(data, "is_template", None, type=bool), event_type=event_type, description=dict_get(data, "description", None, type=str), ) if "jobs" in data: for job in data["jobs"]: _add_job(event, job) return jsonify(event) except KeyError: raise BadRequest("Missing POST parameter") except (NotFound, ValueError): raise BadRequest("Invalid parameter") @blueprint.route("/events/", methods=["PUT"]) @login_required(permission=permissions.EDIT) def modify_event(event_id, current_session): """Modify an event Route: ``/events/`` | 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) @blueprint.route("/events/", methods=["DELETE"]) @login_required(permission=permissions.DELETE) def delete_event(event_id, current_session): """Delete an event Route: ``/events/`` | 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 @blueprint.route("/events//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//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) @blueprint.route("/events//jobs/", methods=["DELETE"]) @login_required(permission=permissions.DELETE) def delete_job(event_id, job_id, current_session): """Delete a Job Route: ``/events//jobs/`` | 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() @blueprint.route("/events//jobs/", methods=["PUT"]) @login_required() def update_job(event_id, job_id, current_session: Session): """Edit Job Route: ``/events//jobs/`` | 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 or "type_id" in data: job.type_ = event_controller.get_job_type(data.get("type", None) or data["type_id"]) 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) @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}) @blueprint.route("/events/jobs/", methods=["GET"]) @login_required() def get_job(job_id, current_session: Session): return jsonify(event_controller.get_job(job_id)) @blueprint.route("/events/jobs//assign", methods=["POST"]) @login_required() def assign_job(job_id, current_session: Session): """Assign / unassign user to the Job Route: ``/events/jobs//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: value = data["value"] user = userController.get_user( data["userid"], deleted=value < 0 ) # allow unassigning deleted users, but not assigning 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) @blueprint.route("/events/jobs//lock", methods=["POST"]) @login_required(permissions.LOCK_JOBS) def lock_job(job_id, current_session: Session): """Lock / unlock the Job Route: ``/events/jobs//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() @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[], transferee?: string}`` 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 or 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_, userController.get_user(transferee) if transferee else None ) for invitee in [userController.get_user(uid) for uid in data["invitees"]] ] ) except (TypeError, KeyError, ValueError, NotFound): raise BadRequest @blueprint.route("/events/invitations", methods=["GET"]) @login_required() def get_invitations(current_session: Session): return jsonify(event_controller.get_invitations(current_session.user_)) @blueprint.route("/events/invitations/", 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 NotFound return jsonify(inv) @blueprint.route("/events/invitations/", 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()