241 lines
6.7 KiB
Python
241 lines
6.7 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,
|
||
)
|
||
|
||
|
||
from datetime import datetime, date
|
||
from flask import Response
|
||
from bson import json_util
|
||
|
||
def list_patchnotes(since: str | None = None):
|
||
"""GET /patchnotes – liefert neue Patchnotes und setzt last_login."""
|
||
|
||
user_id = current_user_id()
|
||
|
||
# Last login aus DB holen
|
||
last_login_doc = last_login_collection.find_one({"user_id": user_id})
|
||
last_login = last_login_doc.get("last_login") if last_login_doc else None
|
||
|
||
# Abfrage definieren (default: alles holen)
|
||
query = {}
|
||
|
||
if since:
|
||
# since validieren
|
||
try:
|
||
since_date = datetime.strptime(since, "%Y-%m-%d").date()
|
||
except ValueError:
|
||
raise ProblemException(
|
||
title="Bad Request",
|
||
detail="since must be in format YYYY-MM-DD",
|
||
status=400,
|
||
)
|
||
|
||
# last_login ebenfalls in date konvertieren, wenn vorhanden
|
||
if last_login:
|
||
if isinstance(last_login, str):
|
||
last_login = datetime.strptime(last_login, "%Y-%m-%d").date()
|
||
elif isinstance(last_login, datetime):
|
||
last_login = last_login.date()
|
||
|
||
# Datumsbereichs-Query
|
||
query = {
|
||
"patch_date": {
|
||
"$gte": last_login if last_login else date.min,
|
||
"$lte": since_date
|
||
}
|
||
}
|
||
|
||
# Patchnotes holen
|
||
docs = list(collection.find(query).sort("patch_date", 1))
|
||
for d in docs:
|
||
d.pop("_id", None)
|
||
if isinstance(d.get("patch_date"), (datetime, date)):
|
||
d["patch_date"] = d["patch_date"].isoformat()
|
||
|
||
# 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,
|
||
)
|
||
|
||
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,
|
||
)
|