225 lines
6.5 KiB
Python
225 lines
6.5 KiB
Python
import logging
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
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 # noqa: E501
|
|
|
|
|
|
def patchnotes_date_get(date): # noqa: E501
|
|
"""patchnotes_date_get
|
|
|
|
returns list of all unseen Patch notes from the user # noqa: E501
|
|
|
|
:param date:
|
|
:type date: str
|
|
|
|
:rtype: Union[List[PatchNotes], Tuple[List[PatchNotes], int], Tuple[List[PatchNotes], int, Dict[str, str]]
|
|
"""
|
|
logging.error(type(date))
|
|
try:
|
|
datetime.strptime(date, '%Y-%m-%d') # 2025-01-0
|
|
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,
|
|
)
|
|
logging.error('date:' + str(date))
|
|
query = {
|
|
'patch_date': {'$gte': date}
|
|
}
|
|
results = []
|
|
for entry in collection.find(query):
|
|
patch_note = PatchNotes.from_dict(entry)
|
|
results.append(patch_note.to_dict())
|
|
return Response(
|
|
json_util.dumps(results),
|
|
mimetype='application/json',
|
|
status=200,
|
|
)
|
|
|
|
|
|
def patchnotes_patch_iddelete(patch_id): # noqa: E501
|
|
"""patchnotes_patch_iddelete
|
|
|
|
deletes one Patch note # noqa: E501
|
|
|
|
:param patch_id:
|
|
: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})
|
|
logging.error(deleted_doc.deleted_count)
|
|
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 patchnotes_patch_idput(patch_id, body): # noqa: E501
|
|
"""patchnotes_patch_idput
|
|
|
|
Updates one Patch # noqa: E501
|
|
|
|
:param body:
|
|
:param patch_id:
|
|
:type patch_id: str
|
|
:type patch_id: str
|
|
|
|
:rtype: Union[None, Tuple[None, int], Tuple[None, 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,
|
|
)
|
|
logging.error(result)
|
|
return Response("Document was updated", status=200, mimetype='application/json')
|
|
|
|
|
|
def patchnotes_post(body):
|
|
"""patchnotes_post
|
|
|
|
creates new Patch notes # noqa: E501
|
|
|
|
:param body:
|
|
:param patch_notes:
|
|
:type patch_notes: dict | bytes
|
|
|
|
:rtype: Union[List[PatchNotes], Tuple[List[PatchNotes], int], Tuple[List[PatchNotes], int, Dict[str, str]]
|
|
"""
|
|
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,
|
|
)
|
|
|
|
logging.error('patch_notes_type:' + str(type(date_obj)))
|
|
logging.error('patch notes' + str(pn))
|
|
|
|
patch_note = {
|
|
'patchID': str(uuid.uuid4()),
|
|
'title': pn.title,
|
|
'changes': pn.changes,
|
|
'version': pn.version,
|
|
'patch_date': date_str
|
|
}
|
|
|
|
logging.error('patch_notes_type:' + str(type(patch_note)))
|
|
logging.error('patch notes' + str(patch_note))
|
|
|
|
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 patchnotes_post(body):
|
|
"""patchnotes_post
|
|
|
|
creates new Patch notes # noqa: E501
|
|
|
|
:param body:
|
|
:param patch_notes:
|
|
:type patch_notes: dict | bytes
|
|
|
|
:rtype: Union[List[PatchNotes], Tuple[List[PatchNotes], int], Tuple[List[PatchNotes], int, Dict[str, str]]
|
|
"""
|
|
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 patchnotes_get(): # noqa: E501
|
|
"""patchnotes_get
|
|
|
|
returns list of all Patch notes # noqa: E501
|
|
|
|
|
|
:rtype: Union[List[PatchNotes], Tuple[List[PatchNotes], int], Tuple[List[PatchNotes], int, Dict[str, str]]
|
|
"""
|
|
results = []
|
|
for entry in collection.find({}):
|
|
patch_note = PatchNotes.from_dict(entry)
|
|
results.append(patch_note.to_dict())
|
|
return Response(
|
|
json_util.dumps(results),
|
|
mimetype='application/json',
|
|
status=200,
|
|
)
|
|
|
|
def cors_options_handler():
|
|
from flask import make_response
|
|
response = make_response('')
|
|
response.headers['Access-Control-Allow-Origin'] = '*'
|
|
response.headers['Access-Control-Allow-Methods'] = 'GET, POST, PUT, DELETE, OPTIONS'
|
|
response.headers['Access-Control-Allow-Headers'] = 'Content-Type, Authorization'
|
|
return response
|
|
|