mirror of
https://gitlab.com/crafty-controller/crafty-4.git
synced 2024-08-30 18:23:09 +00:00
93 lines
2.8 KiB
Python
93 lines
2.8 KiB
Python
import logging
|
|
import datetime
|
|
import typing as t
|
|
from peewee import (
|
|
CharField,
|
|
DoesNotExist,
|
|
AutoField,
|
|
DateTimeField,
|
|
)
|
|
from playhouse.shortcuts import model_to_dict
|
|
|
|
from app.classes.models.base_model import BaseModel
|
|
from app.classes.shared.helpers import Helpers
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# **********************************************************************************
|
|
# Roles Class
|
|
# **********************************************************************************
|
|
class Roles(BaseModel):
|
|
role_id = AutoField()
|
|
created = DateTimeField(default=datetime.datetime.now)
|
|
last_update = DateTimeField(default=datetime.datetime.now)
|
|
role_name = CharField(default="", unique=True, index=True)
|
|
|
|
class Meta:
|
|
table_name = "roles"
|
|
|
|
|
|
# **********************************************************************************
|
|
# Roles Helpers
|
|
# **********************************************************************************
|
|
class HelperRoles:
|
|
def __init__(self, database):
|
|
self.database = database
|
|
|
|
@staticmethod
|
|
def get_all_roles():
|
|
return Roles.select()
|
|
|
|
@staticmethod
|
|
def get_all_role_ids() -> t.List[int]:
|
|
return [role.role_id for role in Roles.select(Roles.role_id).execute()]
|
|
|
|
@staticmethod
|
|
def get_roleid_by_name(role_name):
|
|
try:
|
|
return (Roles.get(Roles.role_name == role_name)).role_id
|
|
except DoesNotExist:
|
|
return None
|
|
|
|
@staticmethod
|
|
def get_role(role_id):
|
|
return model_to_dict(Roles.get(Roles.role_id == role_id))
|
|
|
|
@staticmethod
|
|
def get_role_columns(
|
|
role_id: t.Union[str, int], column_names: t.List[str]
|
|
) -> t.List[t.Any]:
|
|
columns = [getattr(Roles, column) for column in column_names]
|
|
return model_to_dict(
|
|
Roles.select(*columns).where(Roles.role_id == role_id).get(),
|
|
only=columns,
|
|
)
|
|
|
|
@staticmethod
|
|
def get_role_column(role_id: t.Union[str, int], column_name: str) -> t.Any:
|
|
column = getattr(Roles, column_name)
|
|
return getattr(
|
|
Roles.select(column).where(Roles.role_id == role_id).get(), column_name
|
|
)
|
|
|
|
@staticmethod
|
|
def add_role(role_name):
|
|
role_id = Roles.insert(
|
|
{
|
|
Roles.role_name: role_name.lower(),
|
|
Roles.created: Helpers.get_time_as_string(),
|
|
}
|
|
).execute()
|
|
return role_id
|
|
|
|
@staticmethod
|
|
def update_role(role_id, up_data):
|
|
return Roles.update(up_data).where(Roles.role_id == role_id).execute()
|
|
|
|
def remove_role(self, role_id):
|
|
return Roles.delete().where(Roles.role_id == role_id).execute()
|
|
|
|
@staticmethod
|
|
def role_id_exists(role_id) -> bool:
|
|
return Roles.select().where(Roles.role_id == role_id).count() != 0
|