Flaschengeist/flaschengeist
Flaschengeist
/
flaschengeist
7
0
Fork 0
Code Issues 20 Pull Requests Packages Projects Releases 2 Wiki Activity

[events] Allow locking events

This commit is contained in:
Ferdinand Thiessen 2021-11-21 17:58:28 +01:00
parent 51a3a8dfc8
commit 04d5b1e83a
3 changed files with 114 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
flaschengeist/plugins/events/models.py
View File

@ -65,6 +65,7 @@ class Job(db.Model, ModelSerializeMixin):
end: Optional[datetime] = db.Column(UtcDateTime)
type: Union[JobType, int] = db.relationship("JobType")
comment: Optional[str] = db.Column(db.String(256))
locked: bool = db.Column(db.Boolean())
services: list[Service] = db.relationship("Service", back_populates="job_")
required_services: float = db.Column(db.Numeric(precision=4, scale=2, asdecimal=False), nullable=False)

3
flaschengeist/plugins/events/permissions.py
View File

@ -22,4 +22,7 @@ ASSIGN_OTHER = "events_assign_other"
SEE_BACKUP = "events_see_backup"
"""Can see users assigned as backup"""
LOCK_JOBS = "events_lock_jobs"
"""Can lock jobs, no further services can be assigned or unassigned"""
permissions = [value for key, value in globals().items() if not key.startswith("_")]

153
flaschengeist/plugins/events/routes.py
View File

@ -1,9 +1,12 @@
from datetime import datetime, timedelta, timezone
from http.client import NO_CONTENT
from re import template
from flask import request, jsonify
from sqlalchemy import exc
from werkzeug.exceptions import BadRequest, NotFound, Forbidden
from flaschengeist.models.session import Session
from flaschengeist.plugins.events.models import Job
from flaschengeist.utils.decorators import login_required
from flaschengeist.utils.datetime import from_iso_format
from flaschengeist.controller import userController
@ -12,6 +15,21 @@ from . import event_controller, permissions, EventPlugin
from ...utils.HTTP import no_content
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):
@ -239,9 +257,7 @@ def get_events(current_session, year=datetime.now().year, month=datetime.now().m
def _add_job(event, data):
try:
start = from_iso_format(data["start"])
end = None
if "end" in data:
end = from_iso_format(data["end"])
end = dict_get(data, "end", None, type=from_iso_format)
required_services = data["required_services"]
job_type = data["type"]
if isinstance(job_type, dict):
@ -256,7 +272,7 @@ def _add_job(event, data):
required_services,
start,
end,
comment=data.get("comment", None),
comment=dict_get(data, "comment", None, str),
)
@ -276,11 +292,9 @@ def create_event(current_session):
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)
end = dict_get(data, "end", None, type=from_iso_format)
data_type = data["type"]
if isinstance(data_type, dict):
data_type = data["type"]["id"]
@ -293,10 +307,10 @@ def create_event(current_session):
event = event_controller.create_event(
start=start,
end=end,
name=data.get("name", None),
is_template=data.get("is_template", None),
name=dict_get(data, "name", None),
is_template=dict_get(data, "is_template", None),
event_type=event_type,
description=data.get("description", None),
description=dict_get(data, "description", None),
)
if "jobs" in data:
for job in data["jobs"]:
@ -323,15 +337,14 @@ def modify_event(event_id, current_session):
"""
event = event_controller.get_event(event_id)
data = request.get_json()
if "start" in data:
event.start = from_iso_format(data["start"])
if "end" in data:
event.end = from_iso_format(data["end"])
if "description" in data:
event.description = data["description"]
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)
@ -390,19 +403,19 @@ def delete_job(event_id, job_id, current_session):
Returns:
HTTP-no-content or HTTP error
"""
job_slot = event_controller.get_job(job_id, event_id)
event_controller.delete_job(job_slot)
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 or assign user to the Job
"""Edit Job
Route: ``/events/<event_id>/jobs/<job_id>`` | Method: ``PUT``
POST-data: See TS interface for Job or ``{user: {userid: string, value: number}}``
POST-data: See TS interface for Job
Args:
event_id: Identifier of the event
@ -412,37 +425,89 @@ def update_job(event_id, job_id, current_session: Session):
Returns:
JSON encoded Job object or HTTP-error
"""
job = event_controller.get_job(job_id, event_id)
if not current_session.user_.has_permission(permissions.EDIT):
raise Forbidden
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
if value > 0:
event_controller.assign_job(job, user, value)
else:
event_controller.unassign_job(job, user, notify=user != current_session.user_)
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()
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/<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:
HTTP-No-Content 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 no_content()
@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()
# TODO: JobTransfer