[events] Implemented API endpoint for jobs of the current user

flaschengeist/plugins/events/event_controller.py

@@ -1,6 +1,6 @@
 from datetime import datetime, timedelta, timezone
 from enum import IntEnum
-from typing import Optional
+from typing import Optional, Tuple
 
 from werkzeug.exceptions import BadRequest, Conflict, NotFound
 from sqlalchemy.exc import IntegrityError
@@ -225,6 +225,24 @@ def get_job(job_id, event_id=None) -> Job:
     return job
 
 
+def get_jobs(user, start=None, end=None, limit=None, offset=None, descending=None) -> Tuple[int, list[Job]]:
+    query = Job.query.join(Service).filter(Service.user_ == user)
+    if start is not None:
+        query = query.filter(start <= Job.end)
+    if end is not None:
+        query = query.filter(end >= Job.start)
+    if descending is not None:
+        query = query.order_by(Job.start.desc(), Job.type_id_)
+    else:
+        query = query.order_by(Job.start, Job.type_id_)
+    count = query.count()
+    if limit is not None:
+        query = query.limit(limit)
+    if offset is not None:
+        query = query.offset(offset)
+    return count, query.all()
+
+
 def add_job(event, job_type, required_services, start, end=None, comment=None):
     job = Job(
         required_services=required_services,

flaschengeist/plugins/events/models.py

@@ -58,7 +58,6 @@ class Service(db.Model, ModelSerializeMixin):
 
 class Job(db.Model, ModelSerializeMixin):
     __tablename__ = _table_prefix_ + "job"
-    _type_id = db.Column("type_id", Serial, db.ForeignKey(f"{_table_prefix_}job_type.id"), nullable=False)
 
     id: int = db.Column(Serial, primary_key=True)
     start: datetime = db.Column(UtcDateTime, nullable=False)
@@ -73,6 +72,8 @@ class Job(db.Model, ModelSerializeMixin):
 
     event_ = db.relationship("Event", back_populates="jobs")
     event_id_ = db.Column("event_id", Serial, db.ForeignKey(f"{_table_prefix_}event.id"), nullable=False)
+    type_id_ = db.Column("type_id", Serial, db.ForeignKey(f"{_table_prefix_}job_type.id"), nullable=False)
+
     invitations_ = db.relationship("Invitation", cascade="all,delete,delete-orphan", back_populates="job_")
 
     __table_args__ = (UniqueConstraint("type_id", "start", "event_id", name="_type_start_uc"),)

flaschengeist/plugins/events/routes.py

@@ -1,10 +1,8 @@
-from datetime import datetime, timedelta, timezone
 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.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
@@ -195,7 +193,7 @@ def get_event(event_id, current_session):
 def get_events(current_session):
     count, result = event_controller.get_events(
         *get_filter_args(),
-            with_backup=current_session.user_.has_permission(permissions.SEE_BACKUP),
+        with_backup=current_session.user_.has_permission(permissions.SEE_BACKUP),
     )
     return jsonify({"count": count, "result": result})
 
@@ -396,6 +394,13 @@ def update_job(event_id, job_id, current_session: Session):
     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):