Loading portal/api/agenda.py +30 −12 Changes for portal/api/agenda.py: 30 added lines, 12 removed lines. Original line number Diff line number Diff line from flask import request from portal.model import TaigaUserStoryProvider from ..model import AgendaItem, Key, TaigaUserStoryReference Loading @@ -7,27 +9,43 @@ from .blueprint import bp @bp.get("/agendas/-/items") @require_valid(Key) def the_agenda_items() -> list[dict[str, str]]: tasks = sorted( def the_agenda_items() -> list[AgendaItem]: user_stories = sorted( TaigaUserStoryProvider.get_the_one().fetch_user_stories(), key=lambda task: task["kanban_order"], key=lambda story: story["kanban_order"], ) item_names = [ "Tagesordnung", "Gremienberichte", "Posteingang und Kummerkasten", *(task["subject"] for task in tasks), "Sonstiges", ] return [ AgendaItem(name="Tagesordnung"), AgendaItem(name="Gremienberichte"), AgendaItem(name="Posteingang und Kummerkasten"), *( AgendaItem(name=story["subject"], taiga=TaigaUserStoryReference(id=story["id"], status=story["status"])) AgendaItem( name=story["subject"], taiga=TaigaUserStoryReference( id=story["id"], status=story["status"], version=story["version"] ), ) for story in user_stories ), AgendaItem(name="Sonstiges"), ] @bp.patch("/agendas/-/items") @require_valid(Key) def patch_the_agenda_items() -> list[AgendaItem]: relevant_items = [item for item in request.json if "taiga" in item] taiga = TaigaUserStoryProvider.get_the_one() # TODO Parallelize this for item in relevant_items: taiga.update_user_story( item["taiga"]["id"], item["taiga"]["version"], status=item["taiga"]["status"], ) # TODO Add a comment return the_agenda_items() portal/model/TaigaUserStoryProvider.py +16 −6 Changes for portal/model/TaigaUserStoryProvider.py: 16 added lines, 6 removed lines. Original line number Diff line number Diff line from sqlalchemy.orm import Mapped from typing import Any, Self import requests from sqlalchemy.orm import Mapped from portal.model.Base import Base class TaigaUserStoryProvider(Base): """ Fetches a list of "user stories" (i.e., tasks) from the project management tool Taiga. Fetches and changes "user stories" (i.e., tasks) from the project management tool Taiga. """ url: Mapped[str] authorization: Mapped[str] def fetch_user_stories(self) -> list[dict[str, Any]]: return requests.get( self.url, headers={"Authorization": self.authorization} ).json() return requests.get(self.url, headers=self.headers).json() def update_user_story(self, id: int, version: int, **kwargs: Any) -> None: # TODO Maybe build this in a smarter way user_story_url = f"{self.url.split('?')[0]}/{id}" # TODO Maybe return something requests.patch( user_story_url, headers=self.headers, json=kwargs | {"version": version} ) @property def headers(self) -> dict[str, str]: return {"Authorization": self.authorization} @classmethod def get_the_one(cls) -> Self: Loading Loading
portal/api/agenda.py +30 −12 Changes for portal/api/agenda.py: 30 added lines, 12 removed lines. Original line number Diff line number Diff line from flask import request from portal.model import TaigaUserStoryProvider from ..model import AgendaItem, Key, TaigaUserStoryReference Loading @@ -7,27 +9,43 @@ from .blueprint import bp @bp.get("/agendas/-/items") @require_valid(Key) def the_agenda_items() -> list[dict[str, str]]: tasks = sorted( def the_agenda_items() -> list[AgendaItem]: user_stories = sorted( TaigaUserStoryProvider.get_the_one().fetch_user_stories(), key=lambda task: task["kanban_order"], key=lambda story: story["kanban_order"], ) item_names = [ "Tagesordnung", "Gremienberichte", "Posteingang und Kummerkasten", *(task["subject"] for task in tasks), "Sonstiges", ] return [ AgendaItem(name="Tagesordnung"), AgendaItem(name="Gremienberichte"), AgendaItem(name="Posteingang und Kummerkasten"), *( AgendaItem(name=story["subject"], taiga=TaigaUserStoryReference(id=story["id"], status=story["status"])) AgendaItem( name=story["subject"], taiga=TaigaUserStoryReference( id=story["id"], status=story["status"], version=story["version"] ), ) for story in user_stories ), AgendaItem(name="Sonstiges"), ] @bp.patch("/agendas/-/items") @require_valid(Key) def patch_the_agenda_items() -> list[AgendaItem]: relevant_items = [item for item in request.json if "taiga" in item] taiga = TaigaUserStoryProvider.get_the_one() # TODO Parallelize this for item in relevant_items: taiga.update_user_story( item["taiga"]["id"], item["taiga"]["version"], status=item["taiga"]["status"], ) # TODO Add a comment return the_agenda_items()
portal/model/TaigaUserStoryProvider.py +16 −6 Changes for portal/model/TaigaUserStoryProvider.py: 16 added lines, 6 removed lines. Original line number Diff line number Diff line from sqlalchemy.orm import Mapped from typing import Any, Self import requests from sqlalchemy.orm import Mapped from portal.model.Base import Base class TaigaUserStoryProvider(Base): """ Fetches a list of "user stories" (i.e., tasks) from the project management tool Taiga. Fetches and changes "user stories" (i.e., tasks) from the project management tool Taiga. """ url: Mapped[str] authorization: Mapped[str] def fetch_user_stories(self) -> list[dict[str, Any]]: return requests.get( self.url, headers={"Authorization": self.authorization} ).json() return requests.get(self.url, headers=self.headers).json() def update_user_story(self, id: int, version: int, **kwargs: Any) -> None: # TODO Maybe build this in a smarter way user_story_url = f"{self.url.split('?')[0]}/{id}" # TODO Maybe return something requests.patch( user_story_url, headers=self.headers, json=kwargs | {"version": version} ) @property def headers(self) -> dict[str, str]: return {"Authorization": self.authorization} @classmethod def get_the_one(cls) -> Self: Loading