#########################################################################
#
# Copyright (C) 2021 OSGeo
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
#########################################################################
from abc import ABCMeta, abstractmethod
from polymorphic.models import PolymorphicModel
from polymorphic.managers import PolymorphicManager
from django.db import models
from django.utils.translation import ugettext_noop as _
[docs]
class AbstractProcessingTaskManager(PolymorphicManager):
[docs]
def get_real_instances(self):
return super().get_queryset().get_real_instances()
[docs]
class AbstractProcessingTask(PolymorphicModel):
[docs]
objects = AbstractProcessingTaskManager()
[docs]
name = models.CharField(max_length=255, unique=True)
[docs]
is_enabled = models.BooleanField(
default=True, help_text=_("Disabling this Task will make the Processing Workflow to skip it.")
)
[docs]
def __str__(self):
get_icon = lambda arg: "[✓]" if arg else "[✗]"
_enabled_icon = get_icon(self.is_enabled)
return f"{_enabled_icon} {self.name} | <{type(self).__name__}>"
[docs]
class ProcessingWorkflow(models.Model):
[docs]
name = models.CharField(max_length=255, unique=True)
[docs]
processing_tasks = models.ManyToManyField(AbstractProcessingTask, blank=True, through="ProcessingWorkflowTasks")
[docs]
is_enabled = models.BooleanField(
default=True, help_text=_("Disabling this Task will make the Processing Workflow to skip it.")
)
[docs]
def get_tasks(self):
return list(self.processing_tasks.order_by("link_to_workflow__order"))
[docs]
def __str__(self):
get_icon = lambda arg: "[✓]" if arg else "[✗]"
_enabled_icon = get_icon(self.is_enabled)
return f"{_enabled_icon} {self.name}"
[docs]
class ProcessingWorkflowTasks(models.Model):
[docs]
workflow = models.ForeignKey(ProcessingWorkflow, on_delete=models.DO_NOTHING)
[docs]
task = models.ForeignKey(AbstractProcessingTask, related_name="link_to_workflow", on_delete=models.DO_NOTHING)
[docs]
order = models.PositiveIntegerField()
[docs]
def __str__(self):
return f"{self.order} --- Workflow: {self.workflow} -- Task: {self.task}"
# ##################################################################################### #
# Samples
# ##################################################################################### #
[docs]
class SampleProcessingTask(AbstractProcessingTask, metaclass=AbstractProcessingTaskMeta):
[docs]
def execute(self, resource):
print(f"Executing {self.name} against {resource}")