416 lines
12 KiB
Python
416 lines
12 KiB
Python
"""Schedule plugin
|
|
|
|
Provides duty schedule / duty roster functions
|
|
"""
|
|
from datetime import datetime, timedelta, timezone
|
|
from http.client import NO_CONTENT
|
|
from flask import Blueprint, request, jsonify
|
|
from werkzeug.exceptions import BadRequest, NotFound, Forbidden
|
|
|
|
from flaschengeist.plugins import Plugin
|
|
from flaschengeist.models.session import Session
|
|
from flaschengeist.utils.decorators import login_required
|
|
from flaschengeist.utils.datetime import from_iso_format
|
|
from flaschengeist.controller import userController
|
|
|
|
from . import event_controller, permissions
|
|
from . import models
|
|
from ...utils.HTTP import no_content
|
|
|
|
schedule_bp = Blueprint("schedule", __name__, url_prefix="/schedule")
|
|
|
|
|
|
class SchedulePlugin(Plugin):
|
|
models = models
|
|
|
|
def __init__(self, config):
|
|
super().__init__(
|
|
blueprint=schedule_bp,
|
|
permissions=permissions.permissions,
|
|
)
|
|
|
|
|
|
@schedule_bp.route("/event-types", methods=["GET"])
|
|
@schedule_bp.route("/event-types/<int:identifier>", methods=["GET"])
|
|
@login_required()
|
|
def get_event_types(current_session, identifier=None):
|
|
"""Get EventType(s)
|
|
|
|
Route: ``/schedule/event-types`` | Method: ``GET``
|
|
Route: ``/schedule/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)
|
|
|
|
|
|
@schedule_bp.route("/event-types", methods=["POST"])
|
|
@login_required(permission=permissions.EVENT_TYPE)
|
|
def new_event_type(current_session):
|
|
"""Create a new EventType
|
|
|
|
Route: ``/schedule/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)
|
|
|
|
|
|
@schedule_bp.route("/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: ``/schedule/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
|
|
|
|
|
|
@schedule_bp.route("/job-types", methods=["GET"])
|
|
@login_required()
|
|
def get_job_types(current_session):
|
|
"""Get all JobTypes
|
|
|
|
Route: ``/schedule/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)
|
|
|
|
|
|
@schedule_bp.route("/job-types", methods=["POST"])
|
|
@login_required(permission=permissions.JOB_TYPE)
|
|
def new_job_type(current_session):
|
|
"""Create a new JobType
|
|
|
|
Route: ``/schedule/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)
|
|
|
|
|
|
@schedule_bp.route("/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: ``/schedule/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
|
|
|
|
|
|
@schedule_bp.route("/events/<int:event_id>", methods=["GET"])
|
|
@login_required()
|
|
def get_event(event_id, current_session):
|
|
"""Get event by id
|
|
|
|
Route: ``/schedule/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)
|
|
return jsonify(event)
|
|
|
|
|
|
@schedule_bp.route("/events", methods=["GET"])
|
|
@schedule_bp.route("/events/<int:year>/<int:month>", methods=["GET"])
|
|
@schedule_bp.route("/events/<int:year>/<int:month>/<int:day>", methods=["GET"])
|
|
@login_required()
|
|
def get_events(current_session, year=datetime.now().year, month=datetime.now().month, day=None):
|
|
"""Get Event objects for specified date (or month or year),
|
|
if nothing set then events for current month are returned
|
|
|
|
Route: ``/schedule/events[/<year>/<month>[/<int:day>]]`` | Method: ``GET``
|
|
|
|
Args:
|
|
year (int, optional): year to query, defaults to current year
|
|
month (int, optional): month to query (if set), defaults to current month
|
|
day (int, optional): day to query events for (if set)
|
|
current_session: Session sent with Authorization Header
|
|
|
|
Returns:
|
|
JSON encoded list containing events found or HTTP-error
|
|
"""
|
|
try:
|
|
begin = datetime(year=year, month=month, day=1, tzinfo=timezone.utc)
|
|
if day:
|
|
begin += timedelta(days=day - 1)
|
|
end = begin + timedelta(days=1)
|
|
else:
|
|
if month == 12:
|
|
end = datetime(year=year + 1, month=1, day=1, tzinfo=timezone.utc)
|
|
else:
|
|
end = datetime(year=year, month=month + 1, day=1, tzinfo=timezone.utc)
|
|
|
|
events = event_controller.get_events(begin, end)
|
|
return jsonify(events)
|
|
except ValueError:
|
|
raise BadRequest("Invalid date given")
|
|
|
|
|
|
def _add_job(event, data):
|
|
end = None
|
|
try:
|
|
start = from_iso_format(data["start"])
|
|
except (KeyError, ValueError):
|
|
raise BadRequest("Missing POST parameter")
|
|
if "end" in data:
|
|
end = from_iso_format(data["end"])
|
|
|
|
if "required_services" not in data:
|
|
raise BadRequest
|
|
job_type = data["type"]
|
|
if isinstance(job_type, dict):
|
|
job_type = data["type"]["id"]
|
|
job_type = event_controller.get_job_type(job_type)
|
|
event_controller.add_job(event, job_type, data["required_services"], start, end, comment=data.get("comment"))
|
|
|
|
|
|
@schedule_bp.route("/events", methods=["POST"])
|
|
@login_required(permission=permissions.CREATE)
|
|
def create_event(current_session):
|
|
"""Create an new event
|
|
|
|
Route: ``/schedule/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()
|
|
end = data.get("end", None)
|
|
try:
|
|
start = from_iso_format(data["start"])
|
|
if end is not None:
|
|
end = from_iso_format(end)
|
|
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, event_type=event_type, description=data.get("description", None)
|
|
)
|
|
if "jobs" in data:
|
|
for job in data["jobs"]:
|
|
_add_job(event, job)
|
|
return jsonify(event)
|
|
|
|
|
|
@schedule_bp.route("/events/<int:event_id>", methods=["PUT"])
|
|
@login_required(permission=permissions.EDIT)
|
|
def modify_event(event_id, current_session):
|
|
"""Modify an event
|
|
|
|
Route: ``/schedule/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()
|
|
if "start" in data:
|
|
event.start = from_iso_format(data["start"])
|
|
if "description" in data:
|
|
event.description = data["description"]
|
|
if "type" in data:
|
|
event_type = event_controller.get_event_type(data["type"])
|
|
event.type = event_type
|
|
event_controller.update()
|
|
return jsonify(event)
|
|
|
|
|
|
@schedule_bp.route("/events/<int:event_id>", methods=["DELETE"])
|
|
@login_required(permission=permissions.DELETE)
|
|
def delete_event(event_id, current_session):
|
|
"""Delete an event
|
|
|
|
Route: ``/schedule/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
|
|
|
|
|
|
@schedule_bp.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: ``/schedule/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)
|
|
|
|
|
|
@schedule_bp.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_slot = event_controller.get_job(job_id, event_id)
|
|
event_controller.delete_job(job_slot)
|
|
return no_content()
|
|
|
|
|
|
@schedule_bp.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 or assign user to the Job
|
|
|
|
Route: ``/events/<event_id>/jobs/<job_id>`` | Method: ``PUT``
|
|
|
|
POST-data: See TS interface for Job or ``{user: {userid: string, value: number}}``
|
|
|
|
Args:
|
|
event_id: Identifier of the event
|
|
job_id: Identifier of the Job
|
|
current_session: Session sent with Authorization Header
|
|
|
|
Returns:
|
|
JSON encoded Event object or HTTP-error
|
|
"""
|
|
job = event_controller.get_job(job_id, event_id)
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
raise BadRequest
|
|
|
|
if ("user" not in data or len(data) > 1) and not current_session._user.has_permission(permissions.EDIT):
|
|
raise Forbidden
|
|
|
|
if "user" in data:
|
|
try:
|
|
user = userController.get_user(data["user"]["userid"])
|
|
value = data["user"]["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
|
|
event_controller.assign_to_job(job, user, value)
|
|
except (KeyError, ValueError):
|
|
raise BadRequest
|
|
|
|
if "required_services" in data:
|
|
job.required_services = data["required_services"]
|
|
if "type" in data:
|
|
job.type = event_controller.get_job_type(data["type"])
|
|
event_controller.update()
|
|
|
|
return jsonify(job.event_)
|
|
|
|
|
|
# TODO: JobTransfer
|