""" Author: Michel Peltriaux Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany Contact: michel.peltriaux@sgdnord.rlp.de Created on: 15.11.21 """ import uuid from django.contrib.auth.models import User from django.core.exceptions import ObjectDoesNotExist from django.utils import timezone from django.utils.timezone import now from django.db import models, transaction from compensation.settings import COMPENSATION_IDENTIFIER_TEMPLATE, COMPENSATION_IDENTIFIER_LENGTH, \ ECO_ACCOUNT_IDENTIFIER_TEMPLATE, ECO_ACCOUNT_IDENTIFIER_LENGTH from ema.settings import EMA_ACCOUNT_IDENTIFIER_LENGTH, EMA_ACCOUNT_IDENTIFIER_TEMPLATE from intervention.settings import INTERVENTION_IDENTIFIER_LENGTH, INTERVENTION_IDENTIFIER_TEMPLATE from konova.utils import generators from konova.utils.generators import generate_random_string from user.models import UserActionLogEntry, UserAction class UuidModel(models.Model): """ Encapsules identifying via uuid """ id = models.UUIDField( primary_key=True, default=uuid.uuid4, editable=False, ) class Meta: abstract = True class BaseResource(UuidModel): """ A basic resource model, which defines attributes for every derived model """ created = models.ForeignKey( UserActionLogEntry, on_delete=models.SET_NULL, null=True, blank=True, related_name='+' ) modified = models.ForeignKey( UserActionLogEntry, on_delete=models.SET_NULL, null=True, blank=True, related_name='+', help_text="Last modified" ) class Meta: abstract = True def delete(self, using=None, keep_parents=False): """ Base deleting of a resource Args: using (): keep_parents (): Returns: """ try: self.created.delete() except (ObjectDoesNotExist, AttributeError): # Object does not exist anymore - we can skip this pass super().delete() class BaseObject(BaseResource): """ A basic object model, which specifies BaseResource. Mainly used for intervention, compensation, ecoaccount """ identifier = models.CharField(max_length=1000, null=True, blank=True) title = models.CharField(max_length=1000, null=True, blank=True) deleted = models.ForeignKey(UserActionLogEntry, on_delete=models.SET_NULL, null=True, blank=True, related_name='+') comment = models.TextField(null=True, blank=True) log = models.ManyToManyField(UserActionLogEntry, blank=True, help_text="Keeps all user actions of an object", editable=False) class Meta: abstract = True def mark_as_deleted(self, user: User): """ Mark an entry as deleted Does not delete from database but sets a timestamp for being deleted on and which user deleted the object Args: user (User): The performing user Returns: """ if self.deleted: # Nothing to do here return with transaction.atomic(): action = UserActionLogEntry.get_deleted_action(user) self.deleted = action self.log.add(action) self.save() def add_log_entry(self, action: UserAction, user: User, comment: str): """ Wraps adding of UserActionLogEntry to log Args: action (UserAction): The performed UserAction user (User): Performing user comment (str): The optional comment Returns: """ user_action = UserActionLogEntry.objects.create( user=user, action=action, comment=comment ) self.log.add(user_action) def generate_new_identifier(self) -> str: """ Generates a new identifier for the intervention object Returns: str """ from compensation.models import Compensation, EcoAccount from intervention.models import Intervention from ema.models import Ema definitions = { Intervention: { "length": INTERVENTION_IDENTIFIER_LENGTH, "template": INTERVENTION_IDENTIFIER_TEMPLATE, }, Compensation: { "length": COMPENSATION_IDENTIFIER_LENGTH, "template": COMPENSATION_IDENTIFIER_TEMPLATE, }, EcoAccount: { "length": ECO_ACCOUNT_IDENTIFIER_LENGTH, "template": ECO_ACCOUNT_IDENTIFIER_TEMPLATE, }, Ema: { "length": EMA_ACCOUNT_IDENTIFIER_LENGTH, "template": EMA_ACCOUNT_IDENTIFIER_TEMPLATE, }, } if self.__class__ not in definitions: # Not defined, yet. Create fallback identifier for this case return generate_random_string(10) _now = now() curr_month = str(_now.month) curr_year = str(_now.year) rand_str = generate_random_string( length=definitions[self.__class__]["length"], use_numbers=True, use_letters_lc=False, use_letters_uc=True, ) _str = "{}{}-{}".format(curr_month, curr_year, rand_str) return definitions[self.__class__]["template"].format(_str) class RecordableObjectMixin(models.Model): """ Wraps record related fields and functionality """ # Refers to "verzeichnen" recorded = models.OneToOneField( UserActionLogEntry, on_delete=models.SET_NULL, null=True, blank=True, help_text="Holds data on user and timestamp of this action", related_name="+" ) class Meta: abstract = True def set_unrecorded(self, user: User): """ Perform unrecording Args: user (User): Performing user Returns: """ if not self.recorded: return None action = UserActionLogEntry.get_unrecorded_action(user) self.recorded = None self.save() self.log.add(action) return action def set_recorded(self, user: User): """ Perform recording Args: user (User): Performing user Returns: """ if self.recorded: return None action = UserActionLogEntry.get_recorded_action(user) self.recorded = action self.save() self.log.add(action) return action def mark_as_edited(self, performing_user: User): """ In case the object or a related object changed, internal processes need to be started, such as unrecord and uncheck Args: performing_user (User): The user which performed the editing action Returns: """ if self.recorded: self.set_unrecorded(performing_user) class CheckableObjectMixin(models.Model): # Checks - Refers to "Genehmigen" but optional checked = models.OneToOneField( UserActionLogEntry, on_delete=models.SET_NULL, null=True, blank=True, help_text="Holds data on user and timestamp of this action", related_name="+" ) class Meta: abstract = True def set_unchecked(self) -> None: """ Perform unrecording Args: Returns: """ if not self.checked: # Nothing to do return # Do not .delete() the checked attribute! Just set it to None, since a delete() would kill it out of the # log history, which is not what we want! self.checked = None self.save() return None def set_checked(self, user: User) -> UserActionLogEntry: """ Perform checking Args: user (User): Performing user Returns: """ if self.checked: # Nothing to do return action = UserActionLogEntry.get_checked_action(user) self.checked = action self.save() self.log.add(action) return action class ShareableObjectMixin(models.Model): # Users having access on this object users = models.ManyToManyField(User, help_text="Users having access (data shared with)") access_token = models.CharField( max_length=255, null=True, blank=True, help_text="Used for sharing access", ) class Meta: abstract = True def generate_access_token(self, make_unique: bool = False, rec_depth: int = 5): """ Creates a new access token for the data Tokens are not used for identification of a table row. The share logic checks the intervention id as well as the given token. Therefore two different interventions can hold the same access_token without problems. For (possible) future changes to the share logic, the make_unique parameter may be used for checking whether the access_token is already used in any intervention. If so, tokens will be generated as long as a free token can be found. Args: make_unique (bool): Perform check on uniqueness over all intervention entries rec_depth (int): How many tries for generating a free random token (only if make_unique) Returns: """ # Make sure we won't end up in an infinite loop of trying to generate access_tokens rec_depth = rec_depth - 1 if rec_depth < 0 and make_unique: raise RuntimeError( "Access token generating for {} does not seem to find a free random token! Aborted!".format(self.id) ) # Create random token token = generators.generate_random_string(15, True, True, False) # Check dynamically wheter there is another instance of that model, which holds this random access token _model = self._meta.concrete_model token_used_in = _model.objects.filter(access_token=token) # Make sure the token is not used anywhere as access_token, yet. # Make use of QuerySet lazy method for checking if it exists or not. if token_used_in and make_unique: self.generate_access_token(make_unique, rec_depth) else: self.access_token = token self.save() def is_shared_with(self, user: User): """ Access check Checks whether a given user has access to this object Args: user (): Returns: """ return self.users.filter(id=user.id) def share_with(self, user: User): """ Adds user to list of shared access users Args: user (User): The user to be added to the object Returns: """ if not self.is_shared_with(user): self.users.add(user) def share_with_list(self, user_list: list): """ Sets the list of shared access users Args: user_list (list): The users to be added to the object Returns: """ self.users.set(user_list) def update_sharing_user(self, form): """ Adds a new user with shared access to the object Args: form (ShareModalForm): The form holding the data Returns: """ form_data = form.cleaned_data keep_accessing_users = form_data["users"] new_accessing_users = list(form_data["user_select"].values_list("id", flat=True)) accessing_users = keep_accessing_users + new_accessing_users users = User.objects.filter( id__in=accessing_users ) self.share_with_list(users)