185 lines
5.3 KiB
Python
185 lines
5.3 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
|
||
from openapi_server.models.patch_notes import PatchNotes
|
||
|
||
|
||
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 body is None:
|
||
return Response('error: No patch notes provided', status=400)
|
||
pn = PatchNotes.from_dict(body)
|
||
try:
|
||
date_obj = datetime.strptime(pn.patch_date, '%Y-%m-%d').date()
|
||
date_str = date_obj.strftime('%Y-%m-%d')
|
||
except ValueError as e:
|
||
raise ProblemException(
|
||
title="Bad Request",
|
||
detail="The date is invalid. Needs to be in this Format YYYY-MM-DD",
|
||
status=400,
|
||
)
|
||
|
||
patch_note = {
|
||
'patchID': str(uuid.uuid4()),
|
||
'title': pn.title,
|
||
'changes': pn.changes,
|
||
'version': pn.version,
|
||
'patch_date': date_str
|
||
}
|
||
|
||
collection.insert_one(patch_note)
|
||
patch_note_obj = PatchNotes.from_dict(patch_note)
|
||
patch_note = patch_note_obj.to_dict()
|
||
return Response(json_util.dumps(patch_note), mimetype='application/json',
|
||
status=200)
|
||
|
||
|
||
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):
|
||
# 1) validate query param only if present
|
||
if since:
|
||
try:
|
||
datetime.strptime(since, "%Y-%m-%d")
|
||
except ValueError:
|
||
raise ProblemException(
|
||
title="Bad Request",
|
||
detail="The date is invalid. Needs to be in this format YYYY-MM-DD",
|
||
status=400,
|
||
)
|
||
query = {"patch_date": {"$gte": since}}
|
||
else:
|
||
query = {} # no filter -> all notes
|
||
|
||
# 2) fetch from Mongo
|
||
docs = list(collection.find(query).sort("patch_date", 1))
|
||
|
||
# 3) ensure every patch_date is *string* before serialising
|
||
for d in docs:
|
||
if isinstance(d.get("patch_date"), (datetime, date)):
|
||
d["patch_date"] = d["patch_date"].strftime("%Y-%m-%d")
|
||
d.pop("_id", None) # optional: hide Mongo’s internal id
|
||
|
||
# 4) serialise – default=str handles any remaining non-JSON types
|
||
return Response(
|
||
json_util.dumps(docs, default=str),
|
||
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]]
|
||
"""
|
||
pn = PatchNotes.from_dict(body)
|
||
try:
|
||
date_obj = datetime.strptime(pn.patch_date, '%Y-%m-%d').date()
|
||
date_str = date_obj.strftime('%Y-%m-%d')
|
||
except ValueError as e:
|
||
raise ProblemException(
|
||
title="Bad Request",
|
||
detail="The date is invalid. Needs to be in this Format YYYY-MM-DD",
|
||
status=400,
|
||
)
|
||
query_filter = {'patchID': patch_id}
|
||
updated_post = {'$set':
|
||
{
|
||
'title': pn.title,
|
||
'changes': pn.changes,
|
||
'version': pn.version,
|
||
'date': date_str
|
||
}
|
||
}
|
||
result = collection.update_one(query_filter, updated_post)
|
||
if result.modified_count == 0:
|
||
raise ProblemException(
|
||
title="Not found",
|
||
detail="No document has been deleted",
|
||
status=404,
|
||
)
|
||
return Response("Document was updated", status=200, mimetype='application/json')
|