70 lines
2 KiB
Python
70 lines
2 KiB
Python
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()
|
|
else:
|
|
appRole = AppRole(
|
|
user_id=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(id)
|
|
|
|
@staticmethod
|
|
def __insertAppRoleToUser(userId, userRes):
|
|
app_role = AppRole.query.filter_by(user_id=userId).first()
|
|
userRes["traits"]["role_id"] = app_role.role_id if app_role else None
|
|
|
|
return userRes
|