Moving to FastAPI
Config is not properly working with .env file as of now Cost Centre is mostly working, going to work on Accounts now
This commit is contained in:
tanshu committed
1 parent
72044476a8
commit
fe74ef44bf
89 files changed
+1516
-2421
No files matched your search
Whitespace-only changes.
@@ -0,0 +1,85 @@
|
||||
import uuid
|
||||
from sqlalchemy import desc
|
||||
from brewman.models.auth import Client, LoginHistory
|
||||
|
||||
from fastapi import APIRouter
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.put("/{id}") # "Clients"
|
||||
def update(request):
|
||||
item = (
|
||||
request.dbsession.query(Client)
|
||||
.filter(Client.id == uuid.UUID(request.matchdict["id"]))
|
||||
.first()
|
||||
)
|
||||
item.enabled = request.json_body["enabled"]
|
||||
if item.enabled:
|
||||
item.otp = None
|
||||
item.name = request.json_body["name"]
|
||||
transaction.commit()
|
||||
return {}
|
||||
|
||||
|
||||
@router.delete("/{id}") # "Clients"
|
||||
def delete(request):
|
||||
id_ = request.matchdict.get("id", None)
|
||||
if id_ is None:
|
||||
response = Response("Client not found")
|
||||
response.status_int = 500
|
||||
return response
|
||||
|
||||
client = request.dbsession.query(Client).filter(Client.id == uuid.UUID(id_)).first()
|
||||
request.dbsession.execute(
|
||||
LoginHistory.__table__.delete(LoginHistory.client_id == client.id)
|
||||
)
|
||||
request.dbsession.delete(client)
|
||||
transaction.commit()
|
||||
return {}
|
||||
|
||||
|
||||
@router.get("/") # "Clients"
|
||||
async def show_list(l: bool):
|
||||
list_ = request.dbsession.query(Client).order_by(Client.name).all()
|
||||
clients = []
|
||||
for item in list_:
|
||||
last_login = (
|
||||
request.dbsession.query(LoginHistory)
|
||||
.filter(LoginHistory.client_id == item.id)
|
||||
.order_by(desc(LoginHistory.date))
|
||||
.first()
|
||||
)
|
||||
last_login = (
|
||||
"Never"
|
||||
if last_login is None
|
||||
else last_login.date.strftime("%d-%b-%Y %H:%M")
|
||||
)
|
||||
clients.append(
|
||||
{
|
||||
"id": item.id,
|
||||
"code": item.code,
|
||||
"name": item.name,
|
||||
"enabled": item.enabled,
|
||||
"otp": item.otp,
|
||||
"creationDate": item.creation_date.strftime("%d-%b-%Y %H:%M"),
|
||||
"lastLogin": last_login,
|
||||
}
|
||||
)
|
||||
return clients
|
||||
|
||||
|
||||
@router.get("/{id}") # "Clients"
|
||||
def show_id(request):
|
||||
item = (
|
||||
request.dbsession.query(Client)
|
||||
.filter(Client.id == uuid.UUID(request.matchdict["id"]))
|
||||
.first()
|
||||
)
|
||||
return {
|
||||
"id": item.id,
|
||||
"code": item.code,
|
||||
"name": item.name,
|
||||
"enabled": item.enabled,
|
||||
"otp": item.otp,
|
||||
}
|
||||
@@ -0,0 +1,93 @@
|
||||
import uuid
|
||||
from brewman.models.auth import Group, Role
|
||||
|
||||
from fastapi import APIRouter
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.post("/new") # "Users"
|
||||
def save(request):
|
||||
group = Group(request.json_body["name"])
|
||||
request.dbsession.add(group)
|
||||
add_permissions(group, request.json_body["permissions"], request.dbsession)
|
||||
transaction.commit()
|
||||
return group_info(group.id, request.dbsession)
|
||||
|
||||
|
||||
@router.put("/{id}") # "Users"
|
||||
def update(request):
|
||||
id_ = request.matchdict.get("id", None)
|
||||
group = request.dbsession.query(Group).filter(Group.id == uuid.UUID(id_)).one()
|
||||
group.name = request.json_body["name"]
|
||||
add_permissions(group, request.json_body["permissions"], request.dbsession)
|
||||
transaction.commit()
|
||||
return group_info(group.id, request.dbsession)
|
||||
|
||||
|
||||
def add_permissions(group, permissions, dbsession):
|
||||
for permission in permissions:
|
||||
id_ = uuid.UUID(permission["id"])
|
||||
gp = [p for p in group.roles if p.id == id_]
|
||||
gp = None if len(gp) == 0 else gp[0]
|
||||
if permission["enabled"] and gp is None:
|
||||
group.roles.append(dbsession.query(Role).filter(Role.id == id_).one())
|
||||
elif not permission["enabled"] and gp:
|
||||
group.roles.remove(gp)
|
||||
|
||||
|
||||
@router.delete("/{id}") # "Users"
|
||||
def delete(request):
|
||||
id_ = request.matchdict.get("id", None)
|
||||
if id_ is None:
|
||||
response = Response("Group is Null")
|
||||
response.status_int = 500
|
||||
return response
|
||||
else:
|
||||
response = Response("Group deletion not implemented")
|
||||
response.status_int = 500
|
||||
return response
|
||||
|
||||
|
||||
@router.get("/{id}") # "Users"
|
||||
def show_id(request):
|
||||
return group_info(uuid.UUID(request.matchdict.get("id", None)), request.dbsession)
|
||||
|
||||
|
||||
@router.get("/new") # "Users"
|
||||
def show_blank(request):
|
||||
return group_info(None, request.dbsession)
|
||||
|
||||
|
||||
@router.get("/") # "Users"
|
||||
async def show_list(l: bool):
|
||||
list_ = request.dbsession.query(Group).order_by(Group.name).all()
|
||||
|
||||
groups = []
|
||||
for item in list_:
|
||||
group = {"id": item.id, "name": item.name, "permissions": []}
|
||||
for permission in sorted(item.roles, key=lambda p: p.name):
|
||||
group["permissions"].append(permission.name)
|
||||
groups.append(group)
|
||||
return groups
|
||||
|
||||
|
||||
def group_info(id_, dbsession):
|
||||
if id_ is None:
|
||||
group = {"name": "", "permissions": []}
|
||||
for item in dbsession.query(Role).order_by(Role.name).all():
|
||||
group["permissions"].append(
|
||||
{"id": item.id, "name": item.name, "enabled": False}
|
||||
)
|
||||
else:
|
||||
group_object = dbsession.query(Group).filter(Group.id == id_).one()
|
||||
group = {"id": group_object.id, "name": group_object.name, "permissions": []}
|
||||
for item in dbsession.query(Role).order_by(Role.name).all():
|
||||
group["permissions"].append(
|
||||
{
|
||||
"id": item.id,
|
||||
"name": item.name,
|
||||
"enabled": True if item in group_object.roles else False,
|
||||
}
|
||||
)
|
||||
return group
|
||||
@@ -0,0 +1,179 @@
|
||||
import re
|
||||
import uuid
|
||||
from brewman.models.auth import User, Group
|
||||
from brewman.models.validation_exception import ValidationError
|
||||
|
||||
from fastapi import APIRouter
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
class UserView(object):
|
||||
def __init__(self, request):
|
||||
self.request = request
|
||||
self.user = request.authenticated_userid
|
||||
self.permissions = None
|
||||
if self.user is not None:
|
||||
self.user = (
|
||||
request.dbsession.query(User)
|
||||
.filter(User.id == request.authenticated_userid)
|
||||
.one()
|
||||
)
|
||||
|
||||
@router.post("/new") # "Users"
|
||||
def save(self):
|
||||
return self.save_user()
|
||||
|
||||
def save_user(self):
|
||||
user = User(
|
||||
self.request.json_body["name"],
|
||||
self.request.json_body["password"],
|
||||
self.request.json_body["lockedOut"],
|
||||
)
|
||||
self.request.dbsession.add(user)
|
||||
self.add_groups(user, self.request.json_body["groups"])
|
||||
transaction.commit()
|
||||
return self.user_info(user.id)
|
||||
|
||||
@router.put("/{id}") # "Authenticated"
|
||||
def update(self):
|
||||
id_ = self.request.matchdict["id"]
|
||||
p = re.compile(
|
||||
"^[A-Fa-f0-9]{8}-[A-Fa-f0-9]{4}-[A-Fa-f0-9]{4}-[A-Fa-f0-9]{4}-[A-Fa-f0-9]{12}$"
|
||||
)
|
||||
if p.match(id_):
|
||||
user = (
|
||||
self.request.dbsession.query(User)
|
||||
.filter(User.id == uuid.UUID(id_))
|
||||
.one()
|
||||
)
|
||||
else:
|
||||
user = (
|
||||
self.request.dbsession.query(User).filter(User.name.ilike(id_)).first()
|
||||
)
|
||||
return self.update_user(user)
|
||||
|
||||
def update_user(self, user):
|
||||
if user is None:
|
||||
raise ValidationError("User name / id not found")
|
||||
if self.request.has_permission("Users"):
|
||||
user.name = self.request.json_body["name"]
|
||||
user.locked_out = self.request.json_body["lockedOut"]
|
||||
self.add_groups(user, self.request.json_body["groups"])
|
||||
|
||||
if (
|
||||
self.request.json_body["password"] != ""
|
||||
and self.request.json_body["password"] != user.password
|
||||
):
|
||||
user.password = self.request.json_body["password"]
|
||||
transaction.commit()
|
||||
return self.user_info(user.id)
|
||||
|
||||
@router.delete("/{id}") # "Users"
|
||||
def delete(self):
|
||||
id_ = self.request.matchdict.get("id", None)
|
||||
if id_ is None:
|
||||
response = Response("User is Null")
|
||||
response.status_int = 500
|
||||
return response
|
||||
else:
|
||||
response = Response("User deletion not implemented")
|
||||
response.status_int = 500
|
||||
return response
|
||||
|
||||
@router.get("/{id}") # "Authenticated"
|
||||
def show_id(self):
|
||||
id_ = self.request.matchdict["id"]
|
||||
p = re.compile(
|
||||
"^[A-Za-z0-9]{8}-[A-Za-z0-9]{4}-[A-Za-z0-9]{4}-[A-Za-z0-9]{4}-[A-Za-z0-9]{12}$"
|
||||
)
|
||||
if p.match(id_):
|
||||
id_ = uuid.UUID(id_)
|
||||
return self.user_info(id_)
|
||||
|
||||
@router.get("/new") # "Users"
|
||||
def show_blank(self):
|
||||
return self.user_info(None)
|
||||
|
||||
@router.get("/") # "Users"
|
||||
async def show_list(l: bool):
|
||||
list_ = self.request.dbsession.query(User).order_by(User.name).all()
|
||||
users = []
|
||||
for item in list_:
|
||||
user = {
|
||||
"name": item.name,
|
||||
"lockedOut": item.locked_out,
|
||||
"groups": [g.name for g in item.groups],
|
||||
}
|
||||
users.append(user)
|
||||
return users
|
||||
|
||||
@router.get("/") # "Authenticated"
|
||||
async def show_name(n: bool):
|
||||
list_ = (
|
||||
self.request.dbsession.query(User)
|
||||
.filter(User.locked_out == False)
|
||||
.order_by(User.name)
|
||||
.all()
|
||||
)
|
||||
users = [{"name": item.name} for item in list_]
|
||||
return users
|
||||
|
||||
def user_info(self, id_):
|
||||
if id_ is None:
|
||||
account = {"name": "", "lockedOut": False, "lockedOut": False, "groups": []}
|
||||
for item in self.request.dbsession.query(Group).order_by(Group.name).all():
|
||||
account["groups"].append(
|
||||
{"id": item.id, "name": item.name, "enabled": False}
|
||||
)
|
||||
return account
|
||||
|
||||
if isinstance(id_, uuid.UUID):
|
||||
user = self.request.dbsession.query(User).filter(User.id == id_).one()
|
||||
else:
|
||||
user = (
|
||||
self.request.dbsession.query(User).filter(User.name.ilike(id_)).first()
|
||||
)
|
||||
|
||||
if self.request.has_permission("Users"):
|
||||
account = {
|
||||
"id": user.id,
|
||||
"name": user.name,
|
||||
"password": "",
|
||||
"lockedOut": user.locked_out,
|
||||
"groups": [],
|
||||
}
|
||||
for item in self.request.dbsession.query(Group).order_by(Group.name).all():
|
||||
account["groups"].append(
|
||||
{
|
||||
"id": item.id,
|
||||
"name": item.name,
|
||||
"enabled": True if item in user.groups else False,
|
||||
}
|
||||
)
|
||||
elif self.user.id == user.id:
|
||||
account = {
|
||||
"id": user.id,
|
||||
"name": user.name,
|
||||
"password": "",
|
||||
"lockedOut": user.locked_out,
|
||||
"groups": [],
|
||||
}
|
||||
else:
|
||||
response = Response("User can only update his/her password")
|
||||
response.status_int = 500
|
||||
return response
|
||||
return account
|
||||
|
||||
def add_groups(self, user, groups):
|
||||
for group in groups:
|
||||
id_ = uuid.UUID(group["id"])
|
||||
ug = [g for g in user.groups if g.id == id_]
|
||||
ug = None if len(ug) == 0 else ug[0]
|
||||
if group["enabled"] and ug is None:
|
||||
group_object = (
|
||||
self.request.dbsession.query(Group).filter(Group.id == id_).one()
|
||||
)
|
||||
user.groups.append(group_object)
|
||||
elif not group["enabled"] and ug:
|
||||
user.groups.remove(ug)
|
||||
Reference in new issue
Block a user