44e4e4eb42
introduce admin area first poc for connecting the authentik api Co-authored-by: Philipp Rothmann <philipprothmann@posteo.de> Reviewed-on: #2
100 lines
2.4 KiB
Python
100 lines
2.4 KiB
Python
from flask import jsonify, request
|
|
from flask_jwt_extended import get_jwt, jwt_required
|
|
from flask_cors import cross_origin
|
|
from flask_expects_json import expects_json
|
|
|
|
from areas import api_v1
|
|
from helpers import KratosApi
|
|
from helpers.auth_guard import admin_required
|
|
|
|
from .validation import schema, schema_multiple
|
|
from .lit_user_service import UserService
|
|
|
|
|
|
@api_v1.route("/users", methods=["GET"])
|
|
@jwt_required()
|
|
@cross_origin()
|
|
# @admin_required() TODO: not needed as authentik checks permissions?
|
|
def get_users():
|
|
return jsonify(UserService.get_users())
|
|
|
|
|
|
@api_v1.route("/users/<string:id>", methods=["GET"])
|
|
@jwt_required()
|
|
@cross_origin()
|
|
@admin_required()
|
|
def get_user(id):
|
|
res = UserService.get_user(id)
|
|
return jsonify(res)
|
|
|
|
|
|
@api_v1.route("/users", methods=["POST"])
|
|
@jwt_required()
|
|
@cross_origin()
|
|
@expects_json(schema)
|
|
@admin_required()
|
|
def post_user():
|
|
data = request.get_json()
|
|
res = UserService.post_user(data)
|
|
return jsonify(res)
|
|
|
|
|
|
@api_v1.route("/users/<string:id>", methods=["PUT"])
|
|
@jwt_required()
|
|
@cross_origin()
|
|
@expects_json(schema)
|
|
@admin_required()
|
|
def put_user(id):
|
|
data = request.get_json()
|
|
user_id = __get_user_id_from_jwt()
|
|
res = UserService.put_user(id, user_id, data)
|
|
return jsonify(res)
|
|
|
|
|
|
@api_v1.route("/users/<string:id>", methods=["DELETE"])
|
|
@jwt_required()
|
|
@cross_origin()
|
|
@admin_required()
|
|
def delete_user(id):
|
|
res = KratosApi.delete("/identities/{}".format(id))
|
|
if res.status_code == 204:
|
|
UserService.delete_user(id)
|
|
return jsonify(), res.status_code
|
|
return jsonify(res.json()), res.status_code
|
|
|
|
|
|
@api_v1.route("/users-batch", methods=["POST"])
|
|
@jwt_required()
|
|
@cross_origin()
|
|
@expects_json(schema_multiple)
|
|
@admin_required()
|
|
def post_multiple_users():
|
|
"""Expects an array of user JSON schema in request body."""
|
|
data = request.get_json()
|
|
res = UserService.post_multiple_users(data)
|
|
return jsonify(res)
|
|
|
|
|
|
@api_v1.route("/me", methods=["GET"])
|
|
@jwt_required()
|
|
@cross_origin()
|
|
def get_personal_info():
|
|
user_id = __get_user_id_from_jwt()
|
|
res = UserService.get_user(user_id)
|
|
return jsonify(res)
|
|
|
|
|
|
@api_v1.route("/me", methods=["PUT"])
|
|
@jwt_required()
|
|
@cross_origin()
|
|
@expects_json(schema)
|
|
def update_personal_info():
|
|
data = request.get_json()
|
|
user_id = __get_user_id_from_jwt()
|
|
res = UserService.put_user(user_id, user_id, data)
|
|
return jsonify(res)
|
|
|
|
|
|
def __get_user_id_from_jwt():
|
|
claims = get_jwt()
|
|
return claims["user_id"]
|