235 lines
6.6 KiB
Python
235 lines
6.6 KiB
Python
import connexion
|
||
from typing import Dict
|
||
from typing import Tuple
|
||
from typing import Union
|
||
|
||
from openapi_server.models.patch_notes import PatchNotes # noqa: E501
|
||
from openapi_server.models.patch_notes_input import PatchNotesInput # noqa: E501
|
||
from openapi_server import util
|
||
import logging
|
||
import uuid
|
||
from datetime import datetime, date
|
||
|
||
from bson import json_util
|
||
from connexion import ProblemException
|
||
from flask import Response
|
||
from __main__ import collection
|
||
|
||
from openapi_server.__main__ import collection, last_login_collection
|
||
from openapi_server.models.patch_notes import PatchNotes
|
||
from openapi_server.utils.auth import current_user_id
|
||
|
||
|
||
def create_patchnote(body): # noqa: E501
|
||
"""Creates new patch notes
|
||
|
||
# noqa: E501
|
||
|
||
:param body:
|
||
:param patch_notes_input:
|
||
:type patch_notes_input: dict | bytes
|
||
|
||
:rtype: Union[PatchNotes, Tuple[PatchNotes, int], Tuple[PatchNotes, int, Dict[str, str]]
|
||
"""
|
||
patch_notes_input = body
|
||
if not body:
|
||
return Response("No patch notes provided", status=400)
|
||
|
||
pn = PatchNotes.from_dict(body)
|
||
|
||
# --- Datum wahlweise als str oder Date annehmen ---
|
||
if isinstance(pn.patch_date, date):
|
||
date_str = pn.patch_date.isoformat()
|
||
else:
|
||
try:
|
||
date_str = datetime.strptime(pn.patch_date, "%Y-%m-%d").date().isoformat()
|
||
except ValueError:
|
||
raise ProblemException(
|
||
title="Bad Request",
|
||
detail="patch_date must be in format YYYY-MM-DD",
|
||
status=400,
|
||
)
|
||
|
||
patch_doc = {
|
||
"patchID": str(uuid.uuid4()),
|
||
"title": pn.title,
|
||
"changes": pn.changes,
|
||
"version": pn.version,
|
||
"patch_date": date_str,
|
||
"project": pn.project
|
||
}
|
||
|
||
collection.insert_one(patch_doc)
|
||
patch_doc.pop("_id", None) # Mongo-internes Feld ausblenden
|
||
|
||
return Response(
|
||
json_util.dumps(patch_doc),
|
||
mimetype="application/json",
|
||
status=201, # 201 Created
|
||
headers={"Location": f"/api/patchnotes/{patch_doc['patchID']}"},
|
||
)
|
||
|
||
|
||
def delete_patchnote(patch_id): # noqa: E501
|
||
"""Deletes one patch note
|
||
|
||
# noqa: E501
|
||
|
||
:param patch_id: ID of the patch note
|
||
:type patch_id: str
|
||
:type patch_id: str
|
||
|
||
:rtype: Union[None, Tuple[None, int], Tuple[None, int, Dict[str, str]]
|
||
"""
|
||
deleted_doc = collection.delete_one({'patchID': patch_id})
|
||
if deleted_doc.deleted_count == 0:
|
||
raise ProblemException(
|
||
title="Not found",
|
||
detail="No document has been deleted",
|
||
status=404,
|
||
)
|
||
return Response(
|
||
"Document was deleted",
|
||
mimetype='application/json',
|
||
status=200,
|
||
)
|
||
|
||
|
||
def get_patchnote(patch_id): # noqa: E501
|
||
"""Returns one patch note
|
||
|
||
# noqa: E501
|
||
|
||
:param patch_id: ID of the patch note
|
||
:type patch_id: str
|
||
:type patch_id: str
|
||
|
||
:rtype: Union[PatchNotes, Tuple[PatchNotes, int], Tuple[PatchNotes, int, Dict[str, str]]
|
||
"""
|
||
pn = collection.find_one({'patchID': patch_id})
|
||
if pn is None:
|
||
raise ProblemException(
|
||
title="Not found",
|
||
detail="No document has been found",
|
||
status=404,
|
||
)
|
||
return Response(
|
||
json_util.dumps(pn),
|
||
mimetype='application/json',
|
||
status=200,
|
||
)
|
||
|
||
|
||
def list_patchnotes(since: str | None = None):
|
||
"""GET /patchnotes – liefert neue Patchnotes und setzt last_login."""
|
||
|
||
user_id = current_user_id() # 1) aktueller Benutzer
|
||
|
||
if since is None:
|
||
try:
|
||
docs = collection.find({})
|
||
for d in docs:
|
||
d.pop("_id", None)
|
||
if isinstance(d["patch_date"], (datetime, date)):
|
||
d["patch_date"] = d["patch_date"].isoformat()
|
||
except ValueError as e:
|
||
raise ProblemException(
|
||
title="Internal Server Error",
|
||
detail=str(e),
|
||
status=500,
|
||
)
|
||
|
||
# 3) Datum validieren (nur wenn vorhanden)
|
||
if since:
|
||
try:
|
||
datetime.strptime(since, "%Y-%m-%d")
|
||
except ValueError:
|
||
raise ProblemException(
|
||
title="Bad Request",
|
||
detail="since must be YYYY-MM-DD",
|
||
status=400,
|
||
)
|
||
mongo_filter = {"patch_date": {"$gte": since}}
|
||
else:
|
||
mongo_filter = {} # erster Login → alle Notes
|
||
|
||
# 4) Patchnotes abholen
|
||
docs = list(collection.find(mongo_filter).sort("patch_date", 1))
|
||
for d in docs:
|
||
d.pop("_id", None)
|
||
if isinstance(d["patch_date"], (datetime, date)):
|
||
d["patch_date"] = d["patch_date"].isoformat()
|
||
|
||
# 5) Heutiges Datum als neuen last_login speichern
|
||
today = date.today().isoformat()
|
||
last_login_collection.update_one(
|
||
{"user_id": user_id},
|
||
{"$set": {"last_login": today}},
|
||
upsert=True,
|
||
)
|
||
if since == last_login_collection.find_one({"last_login"}):
|
||
return Response(json_util.dumps([]), mimetype="application/json", status=200)
|
||
|
||
return Response(json_util.dumps(docs), mimetype="application/json", status=200)
|
||
|
||
|
||
def update_patchnote(patch_id, body): # noqa: E501
|
||
"""Updates one patch note (full replace)
|
||
|
||
# noqa: E501
|
||
|
||
:param body:
|
||
:param patch_id: ID of the patch note
|
||
:type patch_id: str
|
||
:type patch_id: str
|
||
:param patch_notes_input:
|
||
:type patch_notes_input: dict | bytes
|
||
|
||
:rtype: Union[PatchNotes, Tuple[PatchNotes, int], Tuple[PatchNotes, int, Dict[str, str]]
|
||
"""
|
||
if not body:
|
||
return Response("No patch notes provided", status=400)
|
||
|
||
pn = PatchNotes.from_dict(body)
|
||
|
||
if isinstance(pn.patch_date, date):
|
||
date_str = pn.patch_date.isoformat()
|
||
else:
|
||
try:
|
||
date_str = datetime.strptime(pn.patch_date, "%Y-%m-%d").date().isoformat()
|
||
except ValueError:
|
||
raise ProblemException(
|
||
title="Bad Request",
|
||
detail="patch_date must be in format YYYY-MM-DD",
|
||
status=400,
|
||
)
|
||
|
||
result = collection.update_one(
|
||
{"patchID": patch_id},
|
||
{"$set": {
|
||
"title": pn.title,
|
||
"changes": pn.changes,
|
||
"version": pn.version,
|
||
"patch_date": date_str, # richtiger Feldname!
|
||
"project": pn.project,
|
||
}}
|
||
)
|
||
|
||
if result.matched_count == 0:
|
||
raise ProblemException(
|
||
title="Not found",
|
||
detail="No document has been found",
|
||
status=404,
|
||
)
|
||
|
||
doc = collection.find_one({"patchID": patch_id})
|
||
doc.pop("_id", None)
|
||
|
||
return Response(
|
||
json_util.dumps(doc),
|
||
mimetype="application/json",
|
||
status=200,
|
||
)
|
||
|
||
|