dashboard/areas/users/user_service.py

import copy

from database import db
from areas.apps import AppRole
from helpers import KratosApi


class UserService:
    @staticmethod
    def get_users():
        res = KratosApi.get("/identities").json()
        userList = []
        for r in res:
            userList.append(UserService.__insertAppRoleToUser(r["id"], r))

        return userList

    @staticmethod
    def get_user(id):
        res = KratosApi.get("/identities/{}".format(id)).json()
        return UserService.__insertAppRoleToUser(id, res)

    @staticmethod
    def post_user(data):
        kratos_data = {
            "schema_id": "default",
            "traits": {"email": data["email"], "name": data["name"]},
        }
        res = KratosApi.post("/identities", kratos_data).json()

        appRole = AppRole(
            user_id=res["id"],
            role_id=data["role_id"] if "role_id" in data else None,
            app_id=1,
        )

        db.session.add(appRole)
        db.session.commit()

        return UserService.get_user(res["id"])

    @staticmethod
    def put_user(id, data):
        kratos_data = {
            "schema_id": "default",
            "traits": {"email": data["email"], "name": data["name"]},
        }
        KratosApi.put("/identities/{}".format(id), kratos_data)

        app_role = AppRole.query.filter_by(user_id=id).first()
        if app_role:
            app_role.role_id = data["role_id"] if "role_id" in data else None
            db.session.commit()

        return UserService.get_user(id)

    @staticmethod
    def __insertAppRoleToUser(userId, userRes):
        app_role = AppRole.query.filter_by(user_id=userId).first()
        userRes["traits"]["app_role_id"] = app_role.role_id if app_role else None

        return userRes