chore(access): planning work for centurion user model, audit history and tanancy model

ref: #765 #766
This commit is contained in:
2025-05-18 04:36:52 +09:30
parent 48977aa8e1
commit af0d2043ac
3 changed files with 534 additions and 1 deletions

View File

@ -0,0 +1,247 @@
# import django
import logging
from django.core.exceptions import (
ValidationError,
)
from django.db import models
# from rest_framework.reverse import reverse
from access.models.tenant import Tenant
# from core import exceptions as centurion_exceptions
# from core.mixin.history_save import SaveHistory
class TenancyManager(
models.Manager
):
"""Multi-Tennant Object Manager
This manager specifically caters for the multi-tenancy features of Centurion ERP.
"""
def get_queryset(self):
""" Fetch the data
When the model contains the user data, the query is filtered to their
and the globally defined Tenancy only.
Returns:
(queryset): **super user**: return unfiltered data.
(queryset): **not super user**: return data from the stored unique organizations.
"""
# user = None # When CenturionUser in use
# if hasattr(self.model, 'context'):
# user = self.model.context['user']
# if user:
# tencies = user.get_tenancies(int_list = True)
# if len(tenancies) > 0 and not request.user.is_superuser:
# if hasattr(self.model, 'organization'):
# return super().get_queryset().select_related('organization').filter(
# models.Q(organization__in = tenancies)
# )
# return super().get_queryset().select_related('organization').filter(
# models.Q(organization__in = tenancies)
# )
request = None
if hasattr(self.model, 'context'):
request = self.model.context['request']
if request is not None:
tenancies: list(str()) = []
if request.app_settings.global_organization:
tenancies += [ request.app_settings.global_organization.id ]
if request.user.is_authenticated:
for team in request.tenancy._user_teams:
if team.organization.id in tenancies:
continue
tenancies += [ team.organization.id ]
if len(tenancies) > 0 and not request.user.is_superuser:
if hasattr(self.model, 'organization'):
return super().get_queryset().select_related('organization').filter(
models.Q(organization__in = tenancies)
)
return super().get_queryset().select_related('organization').filter(
models.Q(organization__in = tenancies)
)
return super().get_queryset().select_related('organization')
class TenancyObjectOld:
kb_model_name: str = None
"""Model name to use for KB article linking
This value is derived from `<model>._meta.model_name`. This value should
only be used when there is model inheritence.
"""
_log: logging.Logger = None
def get_log(self):
if self._log is None:
self._log = logging.getLogger('centurion.' + self._meta.app_label)
return self._log
page_layout: list = None
note_basename: str = None
"""URL BaseName for the notes endpoint.
Don't specify the `app_namespace`, use property `app_namespace` above.
"""
def get_page_layout(self):
""" FEtch the page layout"""
return self.page_layout
def get_app_namespace(self) -> str:
"""Fetch the Application namespace if specified.
Returns:
str: Application namespace suffixed with colin `:`
None: No application namespace found.
"""
app_namespace = ''
if self.app_namespace:
app_namespace = self.app_namespace + ':'
return str(app_namespace)
def get_url_kwargs_notes(self) -> dict:
"""Fetch the URL kwargs for model notes
Returns:
dict: notes kwargs required for generating the URL with `reverse`
"""
return {
'model_id': self.id
}
class TenancyAbstractModel(
# TenancyObjectOld,
# models.Model,
):
""" Tenancy Model Abstract class.
This class is for inclusion within **every** model within Centurion ERP.
Provides the required fields, functions and methods for multi tennant objects.
Unless otherwise stated, **no** object within this class may be overridden.
Raises:
ValidationError: User failed to supply organization
"""
objects = TenancyManager()
""" Multi-Tenanant Objects """
class Meta:
abstract = True
def validatate_organization_exists(self):
"""Ensure that the user did provide an organization
Raises:
ValidationError: User failed to supply organization.
"""
if not self:
raise ValidationError(
code = 'required',
message = 'You must provide an organization'
)
id = models.AutoField(
blank=False,
help_text = 'ID of the item',
primary_key=True,
unique=True,
verbose_name = 'ID'
)
organization = models.ForeignKey(
Tenant,
blank = False,
help_text = 'Tenant this belongs to',
null = False,
on_delete = models.CASCADE,
related_name = '+',
validators = [
validatate_organization_exists
],
verbose_name = 'Tenant'
)
is_global = models.BooleanField(
blank = False,
default = False,
help_text = 'Is this a global object?',
null = False,
verbose_name = 'Global Object'
)
model_notes = models.TextField(
blank = True,
default = None, # ToDo: Remove this field
help_text = 'Tid bits of information',
null = True,
verbose_name = 'Notes',
)
def get_tenant(self) -> Tenant:
""" Return the models Tenancy
This model can be safely over-ridden as long as it returns the models
tenancy
"""
return self.organization

View File

@ -10,6 +10,8 @@ from access.models.tenancy import TenancyObject
from core.models.centurion import (
CenturionModel,
CenturionSubModel,
ValidationError,
)
from core.lib.feature_not_used import FeatureNotUsed
@ -130,6 +132,7 @@ class ModelHistoryOld:
# class CenturionAudit(
class ModelHistory(
ModelHistoryOld,
# CenturionModel,
TenancyObject,
):
"""Centurion Audit History
@ -146,6 +149,9 @@ class ModelHistory(
_audit_enabled: bool = False
"""Don't Save audit history for audit history model"""
_notes_enabled: bool = False
"""Don't create notes table for istory model"""
class Meta:
@ -266,3 +272,85 @@ class ModelHistory(
'after'
]
]
class CenturionAuditTemp:
"""Temp class
This class will merge with `CenturionAudit` (ne `ModelHistory`) when the model history has been
refactored to use MetaModels
"""
def clean_fields(self, exclude = None):
if not self.get_model_history():
raise ValidationError(
code = 'did_not_process_history',
message = 'Unable to process the history.'
)
super().clean_fields(exclude = exclude)
def get_model_history(self, model: models.Model) -> bool:
"""Populate fields `self.before` and `self.after`
Pass in the model that changed and this function will read values
`model.before` and `model.after` to populate the history table.
**Note:** Audit history expects all models to call and save to an
attribute `before` `self.__dict__` and after save to an attribute
called `after`. Prior to calling the after, you must refresh from the
database.
Args:
model (models.Model): The model to get the history for
Returns:
True (bool): History fields populated
Fail (bool): History fields not populated
"""
if not hasattr(model, 'before'):
raise ValidationError(
code = 'model_missing_before_data',
message = 'Unable to save model history as the "before" data is missing.'
)
if not hasattr(model, 'after'):
raise ValidationError(
code = 'model_missing_after_data',
message = 'Unable to save model history as the "after" data is missing.'
)
if model.before == model.after:
raise ValidationError(
code = 'before_and_after_same',
message = 'Unable to save model history.The "before" and "after" data is the same.'
)
# loop through before and after and remove from after any fields that are the same.
return None
class AuditMetaModel(
CenturionSubModel,
# CenturionAudit
ModelHistory,
):
class Meta:
abstract = True
proxy = False

View File

@ -1,12 +1,209 @@
from django.db import models
from django.conf import settings
from django.core.exceptions import (
ValidationError
)
from rest_framework.reverse import reverse
from access.models.tenancy_abstract import TenancyAbstractModel
class CenturionMethodstoMigrate(
TenancyAbstractModel,
# models.Model
):
""" Centurion Model
This class exists containing new objects that will move to class `CenturionModel`
when they are ready.
"""
_audit_enabled: bool = True
"""Should this model have audit history kept"""
_is_submodel: bool = False
"""This model a sub-model"""
_notes_enabled: bool = True
"""Should a table for notes be created for this model"""
context: dict = {
'logger': None,
'request': None,
}
""" Model Context
**ToDo:** Drop request and add user when user model refactored with tenancy.
Context for actions within the model.
"""
def clean_fields(self, exclude=None) -> None:
super().clean_fields(exclude = exclude)
def delete(self, using = None, keep_parents = _is_submodel):
"""Delete Centurion Model
If a model has `_audit_enabled = True`, audit history is populated and
ready to be saved by the audit system (save signal.).
Args:
using (_type_, optional): _description_. Defaults to None.
keep_parents (bool, optional): Keep parent models. Defaults to the
value if is_submodel so as not to delete parent models.
"""
if self._audit_enabled:
self._after = {}
self._before = self.__class__.objects.get( id = self.id ).get_audit_values()
super().delete(using = using, keep_parents = keep_parents)
def full_clean(self, exclude = None, validate_unique = True, validate_constraints = True) -> None:
super().full_clean(
exclude = exclude,
validate_unique = validate_unique,
validate_constraints = validate_constraints
)
def get_audit_values(self) -> dict:
"""Retrieve the field Values
Currently ensures only fields are present.
**ToDo:** Update so the dict that it returns is a dict of dict where each dict
is named after the actual models the fields come from and it contains
only it's fields.
Returns:
dict: Model fields
"""
if self.id is None:
return {}
data = self.__dict__.copy()
clean_data: dict = {}
for name, data in data.items():
for field in self._meta.fields:
if name == field.name:
clean_data.update({
name: data
})
break
return data
def get_after(self) -> dict:
"""Audit Data After Change
Returns:
dict: All model fields after the data changed
"""
return self._after
def get_before(self) -> dict:
"""Audit Data Before Change
Returns:
dict: All model fields before the data changed
"""
return self._before
def get_url( self, relative: bool = False, api_version: int = 2 ) -> str:
"""Return the models API URL
Args:
relative (bool, optional): Return the relative URL for the model. Defaults to False.
api_version (int, optional): API Version to use. Defaults to `2``.
Returns:
str: API URL for the model
"""
namespace = f'v{api_version}'
url_basename = f'{namespace}:_api_{self._meta.model_name}-detail'
url = reverse( viewname = url_basename, kwargs = { 'pk': self.id } )
if relative:
url = settings.SITE_URL + url
return url
def save(self, force_insert = False, force_update = False, using = None, update_fields = None):
"""Save Centurion Model
This Save ensures that `full_clean()` is called so that prior to the
model being saved to the database, it is valid.
If a model has `_audit_enabled = True`, audit history is populated and
ready to be saved by the audit system (save signal.).
"""
self.full_clean(
exclude = None,
validate_unique = True,
validate_constraints = True
)
if self._audit_enabled:
self._after = self.get_audit_values()
self._before = {}
if self.id:
self._before = self.__class__.objects.get( id = self.id ).get_audit_values()
super().save(force_insert=force_insert, force_update=force_update, using=using, update_fields=update_fields)
def validate_constraints(self, exclude = None) -> None:
super().validate_constraints(exclude = exclude)
class CenturionModel(
models.Model
# TenancyAbstractModel,
# models.Model
):
@ -42,6 +239,7 @@ class CenturionModel(
return f'{self._meta.object_name}AuditHistory'
class CenturionSubModel(
CenturionModel
):