flask-demo-api/flask_demo_api/routers/key.py

78 lines
2.0 KiB
Python

from dishka.integrations.flask import FromDishka, inject
from flask import Blueprint, abort, jsonify, request
from flask_demo_api.protocols.models import KeyDTO
from flask_demo_api.usecase.add import PostKey
from flask_demo_api.usecase.delete import DelKey
from flask_demo_api.usecase.get import GetKey
from flask_demo_api.usecase.put import PutKey
key_bp = Blueprint("key_bp", __name__)
@key_bp.route("/", methods=["POST"])
@inject
def past_key(usecase: FromDishka[PostKey]):
json_data = request.get_json()
if json_data:
result = usecase(
request=KeyDTO(key=json_data.get("key"), val=json_data.get("val"))
)
return (
jsonify({"message": "Ok", "data": result}),
201,
)
else:
return jsonify({"message": "No JSON data received"}), 400
@key_bp.route("/", methods=["GET"])
@inject
def get_key(usecase: FromDishka[GetKey]):
request_data = request.args.get("key")
if request_data:
result = usecase(request=KeyDTO(key=request_data))
if result is None:
abort(404, "Key not found")
return (
jsonify({"message": "Ok", "data": result}),
200,
)
else:
abort(400, "Invalid GET parameters")
@key_bp.route("/", methods=["PUT"])
@inject
def put_key(usecase: FromDishka[PutKey]):
json_data = request.get_json()
if json_data:
result = usecase(
request=KeyDTO(key=json_data.get("key"), val=json_data.get("val"))
)
return (
jsonify({"message": "Updated", "data": result}),
200,
)
else:
return jsonify({"message": "No JSON data received"}), 400
@key_bp.route("/", methods=["DELETE"])
@inject
def delete_key(usecase: FromDishka[DelKey]):
json_data = request.get_json()
if json_data:
usecase(request=KeyDTO(key=json_data.get("key"), val=json_data.get("val")))
return (
jsonify({"message": "Deleted"}),
200,
)
else:
return jsonify({"message": "No JSON data received"}), 400