<commit_before>import hashlib
import typing
import uuid

from os import path
from datetime import datetime
from collections import namedtuple

from flask_restful import fields as flask_fields

from sqlalchemy import case
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy import func, select, table, column, exists
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.orm import validates, aliased
from sqlalchemy.sql.schema import CheckConstraint

from cloudify.constants import MGMTWORKER_QUEUE
from cloudify.models_states import (AgentState,
                                    SnapshotState,
                                    ExecutionState,
                                    VisibilityState,
                                    DeploymentModificationState,
                                    DeploymentState)
from cloudify.cryptography_utils import decrypt
from dsl_parser.constants import WORKFLOW_PLUGINS_TO_INSTALL
from dsl_parser.constraints import extract_constraints, validate_input_value
from dsl_parser import exceptions as dsl_exceptions

from manager_rest import config, manager_exceptions
from manager_rest.rest.responses import Workflow, Label
from manager_rest.utils import (get_rrule,
                                classproperty,
                                files_in_folder)
from manager_rest.deployment_update.constants import ACTION_TYPES, ENTITY_TYPES
from manager_rest.constants import (FILE_SERVER_PLUGINS_FOLDER,
                                    FILE_SERVER_RESOURCES_FOLDER,
                                    AUDIT_OPERATIONS)

from .models_base import (
    CreatedAtMixin,
    db,
    JSONString,
    UTCDateTime,
)
from .management_models import User, Manager
from .resource_models_base import SQLResourceBase, SQLModelBase
from .relationships import (
    foreign_key,
    one_to_many_relationship,
    many_to_many_relationship
)
from manager_rest.storage.storage_manager import get_storage_manager


if typing.TYPE_CHECKING:
    from manager_rest.resource_manager import ResourceManager
    from manager_rest.storage.storage_manager import SQLStorageManager


RELATIONSHIP = 'relationship'
NODE = 'node'


# region Top Level Resources

class Blueprint(CreatedAtMixin, SQLResourceBase):
    __tablename__ = 'blueprints'
    __table_args__ = (
        db.Index(
            'blueprints_id__tenant_id_idx',
            'id', '_tenant_id',
            unique=True
        ),
    )

    skipped_fields = dict(
        SQLResourceBase.skipped_fields,
        v1=['main_file_name', 'description']
    )

    main_file_name = db.Column(db.Text)
    plan = db.Column(db.PickleType(protocol=2))
    updated_at = db.Column(UTCDateTime)
    description = db.Column(db.Text)
    is_hidden = db.Column(db.Boolean, nullable=False, default=False)
    state = db.Column(db.Text)
    error = db.Column(db.Text)
    error_traceback = db.Column(db.Text)
    _upload_execution_fk = foreign_key('executions._storage_id',
                                       nullable=True,
                                       ondelete='SET NULL',
                                       deferrable=True,
                                       initially='DEFERRED',
                                       use_alter=True)

    @classproperty
    def labels_model(cls):
        return BlueprintLabel

    @declared_attr
    def labels(cls):
        # labels are defined as `backref` in DeploymentsLabel model
        return None

    @declared_attr
    def upload_execution(cls):
        return db.relationship('Execution',
                               foreign_keys=[cls._upload_execution_fk],
                               cascade='all, delete')

    @classproperty
    def response_fields(cls):
        fields = super(Blueprint, cls).response_fields
        fields['labels'] = flask_fields.List(
            flask_fields.Nested(Label.resource_fields))
        fields['upload_execution'] = flask_fields.Nested(
            Execution.resource_fields)
        return fields

    @classproperty
    def allowed_filter_attrs(cls):
        return ['created_by', 'state', 'id', 'tenant_name']

    def to_response(self, include=None, **kwargs):
        include = include or self.response_fields
        blueprint_dict = super(Blueprint, self).to_response(include, **kwargs)
        if 'labels' in include:
            blueprint_dict['labels'] = self.list_labels(self.labels)
        return blueprint_dict


class Snapshot(CreatedAtMixin, SQLResourceBase):
    __tablename__ = 'snapshots'
    __table_args__ = (
        db.Index(
            'snapshots_id__tenant_id_idx',
            'id', '_tenant_id',
            unique=True
        ),
    )

    status = db.Column(db.Enum(*SnapshotState.STATES, name='snapshot_status'))
    error = db.Column(db.Text)


class Plugin(SQLResourceBase):
    __tablename__ = 'plugins'
    __table_args__ = (
        db.Index(
            'plugins_name_version__tenant_id_idx',
            'package_name', 'package_version', '_tenant_id', 'distribution',
            'distribution_release', 'distribution_version',
            unique=True
        ),
    )
    _extra_fields = {'installation_state': flask_fields.Raw}

    archive_name = db.Column(db.Text, nullable=False, index=True)
    distribution = db.Column(db.Text)
    distribution_release = db.Column(db.Text)
    distribution_version = db.Column(db.Text)
    excluded_wheels = db.Column(db.PickleType(protocol=2))
    package_name = db.Column(db.Text, nullable=False, index=True)
    package_source = db.Column(db.Text)
    package_version = db.Column(db.Text)
    supported_platform = db.Column(db.PickleType(protocol=2))
    supported_py_versions = db.Column(db.PickleType(protocol=2))
    uploaded_at = db.Column(UTCDateTime, nullable=False, index=True)
    wheels = db.Column(db.PickleType(protocol=2), nullable=False)
    title = db.Column(db.Text)
    blueprint_labels = db.Column(JSONString)
    labels = db.Column(JSONString)
    resource_tags = db.Column(JSONString)

    def yaml_file_path(self):
        plugin_dir = path.join(config.instance.file_server_root,
                               FILE_SERVER_PLUGINS_FOLDER,
                               self.id)
        if not path.isdir(plugin_dir):
            return None
        yaml_files = files_in_folder(plugin_dir, '*.yaml')
        return yaml_files[0] if yaml_files else None

    def _yaml_file_name(self):
        yaml_path = self.yaml_file_path()
        return path.basename(yaml_path) if yaml_path else ''

    @property
    def file_server_path(self):
        file_name = self._yaml_file_name()
        if not file_name:
            return ''
        return path.join(FILE_SERVER_RESOURCES_FOLDER,
                         FILE_SERVER_PLUGINS_FOLDER,
                         self.id,
                         file_name)

    @property
    def yaml_url_path(self):
        if not self._yaml_file_name():
            return ''
        return 'plugin:{0}?version={1}&distribution={2}'.format(
            self.package_name,
            self.package_version,
            self.distribution
        )

    @classproperty
    def response_fields(cls):
        fields = super(Plugin, cls).response_fields
        fields['file_server_path'] = flask_fields.String
        fields['yaml_url_path'] = flask_fields.String
        return fields

    def to_response(self, include=None, get_data=False, **kwargs):
        include = include or self.response_fields
        plugin_dict = super(Plugin, self).to_response(include, **kwargs)
        if not get_data:
            plugin_dict['file_server_path'] = ''
        if 'installation_state' in plugin_dict \
                and 'installation_state' in include:
            plugin_dict['installation_state'] = [
                s.to_dict() for s in plugin_dict['installation_state']]
        return plugin_dict

    @declared_attr
    def installation_state(cls):
        return db.relationship('_PluginState', cascade='delete',
                               passive_deletes=True)

    def check_unique_query(self):
        query = self.__class__.query \
            .filter(self.__class__.package_name == self.package_name) \
            .filter(self.__class__.archive_name == self.archive_name)
        if self.visibility != VisibilityState.GLOBAL:
            tenant_or_global_filter = db.or_(
                self.__class__.tenant == self.tenant,
                self.__class__.visibility == VisibilityState.GLOBAL
            )
            query = query.filter(tenant_or_global_filter)
        return query


class _PluginState(SQLModelBase):
    __tablename__ = 'plugins_states'
    _extra_fields = {'manager': flask_fields.Raw}

    _storage_id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    _plugin_fk = foreign_key(Plugin._storage_id)
    _manager_fk = foreign_key('managers.id', nullable=True)
    _agent_fk = foreign_key('agents._storage_id', nullable=True)
    state = db.Column(db.Text)
    error = db.Column(db.Text)

    @classmethod
    def unique_id(cls):
        return '_storage_id'

    def __repr__(self):
        return '<PluginState plugin={0} state={1}>'.format(
            self._plugin_fk, self.state)

    def to_dict(self, suppress_error=False):
        rv = super(_PluginState, self).to_dict(suppress_error)
        rv['manager'] = self.manager.hostname if self.manager else None
        rv['agent'] = self.agent.name if self.agent else None
        return rv

    @declared_attr
    def manager(cls):
        return db.relationship('Manager', lazy='joined')

    @declared_attr
    def agent(cls):
        return db.relationship('Agent', lazy='joined')


class Secret(CreatedAtMixin, SQLResourceBase):
    __tablename__ = 'secrets'
    __table_args__ = (
        db.Index(
            'secrets_id_tenant_id_idx',
            'id', '_tenant_id',
            unique=True
        ),
    )

    value = db.Column(db.Text)
    updated_at = db.Column(UTCDateTime)
    is_hidden_value = db.Column(db.Boolean, nullable=False, default=False)

    @hybrid_property
    def key(self):
        return self.id

    @classproperty
    def resource_fields(cls):
        fields = super(Secret, cls).resource_fields
        fields['key'] = fields.pop('id')
        return fields

    @classproperty
    def allowed_filter_attrs(cls):
        return ['key']


class Site(CreatedAtMixin, SQLResourceBase):
    __tablename__ = 'sites'
    __table_args__ = (
        db.Index(
            'site_name__tenant_id_idx',
            'name', '_tenant_id',
            unique=True
        ),
    )

    name = db.Column(db.Text, nullable=False)
    latitude = db.Column(db.Float, nullable=True)
    longitude = db.Column(db.Float, nullable=True)

    _extra_fields = {
        'location': flask_fields.String
    }

    @property
    def location(self):
        if (self.latitude is None) or (self.longitude is None):
            return None
        return "{0}, {1}".format(self.latitude, self.longitude)

    @classproperty
    def response_fields(cls):
        if not hasattr(cls, '_cached_site_fields'):
            fields = super(Site, cls).response_fields
            fields.pop('id')
            cls._cached_site_fields = fields
        return cls._cached_site_fields

# endregion

# region Derived Resources


class Deployment(CreatedAtMixin, SQLResourceBase):
    __tablename__ = 'deployments'
    __table_args__ = (
        db.Index(
            'deployments__site_fk_visibility_idx',
            '_blueprint_fk', '_site_fk', 'visibility', '_tenant_id'
        ),
        db.Index(
            'deployments_id__tenant_id_idx',
            'id', '_tenant_id',
            unique=True
        ),
        db.Index(
            'deployments__latest_execution_fk_idx',
            '_latest_execution_fk',
            unique=True
        ),
    )
    _extra_fields = {
        'workdir_zip': flask_fields.String,
    }
    skipped_fields = dict(
        SQLResourceBase.skipped_fields,
        v1=['scaling_groups'],
        v2=['scaling_groups']
    )

    # Can we skip check_unique because it was checked in group dep. creation
    guaranteed_unique = False

    # This will be overridden when a workdir is being retrieved
    workdir_zip = None

    description = db.Column(db.Text)
    inputs = db.Column(db.PickleType(protocol=2))
    groups = db.Column(db.PickleType(protocol=2))
    permalink = db.Column(db.Text)
    policy_triggers = db.Column(db.PickleType(protocol=2))
    policy_types = db.Column(db.PickleType(protocol=2))
    outputs = db.Column(db.PickleType(protocol=2, comparator=lambda *a: False))
    capabilities = db.Column(db.PickleType(
        protocol=2, comparator=lambda *a: False))
    scaling_groups = db.Column(db.PickleType(protocol=2))
    updated_at = db.Column(UTCDateTime)
    workflows = db.Column(db.PickleType(
        protocol=2, comparator=lambda *a: False))
    runtime_only_evaluation = db.Column(db.Boolean, default=False)
    installation_status = db.Column(db.Enum(
        DeploymentState.ACTIVE,
        DeploymentState.INACTIVE,
        name="installation_status"
    ))
    deployment_status = db.Column(db.Enum(
            DeploymentState.GOOD,
            DeploymentState.IN_PROGRESS,
            DeploymentState.REQUIRE_ATTENTION,
            name='deployment_status'
        )
    )
    sub_services_status = db.Column(db.Enum(
            DeploymentState.GOOD,
            DeploymentState.IN_PROGRESS,
            DeploymentState.REQUIRE_ATTENTION,
            name='deployment_status'
        ))
    sub_environments_status = db.Column(db.Enum(
            DeploymentState.GOOD,
            DeploymentState.IN_PROGRESS,
            DeploymentState.REQUIRE_ATTENTION,
            name='deployment_status'
        ))
    sub_services_count = db.Column(
        db.Integer, nullable=False, server_default='0')
    sub_environments_count = db.Column(
        db.Integer, nullable=False, server_default='0')
    display_name = db.Column(
        db.Text, nullable=False, index=True,
        default=lambda ctx: ctx.get_current_parameters().get('id')
    )
    resource_tags = db.Column(JSONString)
    _blueprint_fk = foreign_key(Blueprint._storage_id)
    _site_fk = foreign_key(Site._storage_id,
                           nullable=True,
                           ondelete='SET NULL')
    _create_execution_fk = foreign_key('executions._storage_id',
                                       nullable=True,
                                       ondelete='SET NULL',
                                       deferrable=True,
                                       initially='DEFERRED',
                                       use_alter=True)

    _latest_execution_fk = foreign_key('executions._storage_id',
                                       nullable=True,
                                       ondelete='SET NULL',
                                       deferrable=True,
                                       initially='DEFERRED',
                                       use_alter=True,
                                       unique=True)

    deployment_group_id = association_proxy('deployment_groups', 'id')
    latest_execution_finished_operations = association_proxy(
        'latest_execution', 'finished_operations')
    latest_execution_total_operations = association_proxy(
        'latest_execution', 'total_operations')

    @classproperty
    def autoload_relationships(cls):
        return [
            cls.create_execution,
            cls.latest_execution,
            cls.deployment_groups,
            cls.labels,
            cls.schedules,
        ]

    @classproperty
    def labels_model(cls):
        return DeploymentLabel

    @declared_attr
    def create_execution(cls):
        """The create-deployment-environment execution for this deployment"""
        return db.relationship('Execution',
                               foreign_keys=[cls._create_execution_fk],
                               cascade='all, delete',
                               post_update=True)

    @declared_attr
    def latest_execution(cls):
        return db.relationship('Execution',
                               foreign_keys=[cls._latest_execution_fk],
                               cascade='all, delete',
                               post_update=True)

    @declared_attr
    def blueprint(cls):
        return one_to_many_relationship(cls, Blueprint, cls._blueprint_fk)

    blueprint_id = association_proxy('blueprint', 'id')

    @declared_attr
    def site(cls):
        return one_to_many_relationship(cls, Site, cls._site_fk, cascade=False)

    site_name = association_proxy('site', 'name')
    latest_execution_status = association_proxy('latest_execution', 'status')

    @classproperty
    def response_fields(cls):
        if not hasattr(cls, '_cached_deployment_fields'):
            fields = super(Deployment, cls).response_fields
            fields['labels'] = flask_fields.List(
                flask_fields.Nested(Label.resource_fields))
            fields.pop('deployment_group_id', None)
            fields['workflows'] = flask_fields.List(
                flask_fields.Nested(Workflow.resource_fields)
            )
            fields['schedules'] = flask_fields.List(
                flask_fields.Nested(ExecutionSchedule.resource_fields))
            fields['deployment_groups'] = \
                flask_fields.List(flask_fields.String)
            fields['latest_execution_status'] = flask_fields.String()
            fields['latest_execution_total_operations'] = \
                flask_fields.Integer()
            fields['latest_execution_finished_operations'] = \
                flask_fields.Integer()
            fields['has_sub_deployments'] = flask_fields.Boolean()
            fields['create_execution'] = flask_fields.String()
            fields['latest_execution'] = flask_fields.String()
            cls._cached_deployment_fields = fields
        return cls._cached_deployment_fields

    @classproperty
    def allowed_filter_attrs(cls):
        return ['blueprint_id', 'created_by', 'site_name', 'schedules',
                'tenant_name', 'display_name', 'installation_status']

    def to_response(self, include=None, **kwargs):
        include = include or self.response_fields
        dep_dict = super(Deployment, self).to_response(
            include=include, **kwargs)
        if 'workflows' in include:
            dep_dict['workflows'] = self._list_workflows(self.workflows)
        if 'labels' in include:
            dep_dict['labels'] = self.list_labels(self.labels)
        if 'deployment_groups' in include:
            dep_dict['deployment_groups'] = \
                [g.id for g in self.deployment_groups]
        if 'latest_execution_status' in include:
            dep_dict['latest_execution_status'] = \
                DeploymentState.EXECUTION_STATES_SUMMARY.get(
                    self.latest_execution_status)
        if 'installation_status' in include:
            if not dep_dict.get('installation_status'):
                dep_dict['installation_status'] = DeploymentState.INACTIVE
        if 'latest_execution_total_operations' in include:
            dep_dict['latest_execution_total_operations'] = \
                self.latest_execution_total_operations
        if 'latest_execution_finished_operations' in include:
            dep_dict['latest_execution_finished_operations'] = \
                self.latest_execution_finished_operations
        if 'create_execution' in include:
            dep_dict['create_execution'] = \
                self.create_execution.id if self.create_execution else None
        if 'latest_execution' in include:
            dep_dict['latest_execution'] = \
                self.latest_execution.id if self.latest_execution else None
        return dep_dict

    @staticmethod
    def _list_workflows(deployment_workflows):
        if deployment_workflows is None:
            return None

        return [Workflow(name=wf_name,
                         created_at=None,
                         plugin=wf.get('plugin', ''),
                         operation=wf.get('operation', ''),
                         parameters=wf.get('parameters', dict()),
                         is_cascading=wf.get('is_cascading', False))
                for wf_name, wf in deployment_workflows.items()]

    @classmethod
    def compare_statuses(
                cls, *statuses: typing.Optional[str]
            ) -> typing.Optional[str]:
        """Unify multiple DeploymentStates into a single state.

        Choose the "worst" possible outcome based on the given states,
        ie. if there's one that requires attention, then the overall status
        is also "requires attention".
        """
        if not statuses:
            return None
        importance = {
            DeploymentState.GOOD: 1,
            DeploymentState.IN_PROGRESS: 2,
            DeploymentState.REQUIRE_ATTENTION: 3
        }
        return max(statuses, key=lambda st: importance.get(st, 0))

    def evaluate_sub_deployments_statuses(self):
        """
        Evaluate the deployment statuses per deployment using the following
        three statuses
        1. sub_environments_status
        2. sub_services_status
        3. deployment_status
        This is useful and prerequisite before propagate the source
        deployments statuses to the target
        :return: Tuple of end result of `sub_environments_status`
         & `sub_services_status`
        :rtype Tuple
        """
        _sub_environments_status = self.sub_environments_status
        _sub_services_status = self.sub_services_status
        if self.is_environment:
            _sub_environments_status = \
                self.compare_statuses(
                    self.sub_environments_status,
                    self.deployment_status
                )

        else:
            _sub_services_status = \
                self.compare_statuses(
                    self.sub_services_status,
                    self.deployment_status
                )
        return _sub_services_status, _sub_environments_status

    def evaluate_deployment_status(self, exclude_sub_deployments=False):
        """
        Evaluate the overall deployment status based on installation status
        and latest execution object
        :return: deployment_status: Overall deployment status
        """
        deployment_status = self.decide_deployment_status(
            self.latest_execution_status,
            self.installation_status,
            self.sub_services_status,
            self.sub_environments_status
        )
        self.deployment_status = deployment_status
        return deployment_status

    @classmethod
    def decide_deployment_status(
        cls,
        latest_execution_status,
        installation_status,
        sub_services_status,
        sub_environments_status,
    ):
        latest_status = DeploymentState.EXECUTION_STATES_SUMMARY.get(
            latest_execution_status)
        if latest_status == DeploymentState.IN_PROGRESS:
            deployment_status = DeploymentState.IN_PROGRESS
        elif latest_status == DeploymentState.FAILED \
                or installation_status == DeploymentState.INACTIVE:
            deployment_status = DeploymentState.REQUIRE_ATTENTION
        else:
            deployment_status = DeploymentState.GOOD

        # Check whether or not deployment has services or environments
        # attached to it, so that we can consider that while evaluating the
        # deployment status
        if sub_services_status:
            deployment_status = cls.compare_statuses(
                sub_services_status,
                deployment_status
            )
        if sub_environments_status:
            deployment_status = cls.compare_statuses(
                sub_environments_status,
                deployment_status
            )
        return deployment_status

    @property
    def is_environment(self):
        target_key = '<KEY>'
        target_value = 'environment'
        for label in self.labels:
            if label.key == target_key and label.value == target_value:
                return True
        return False

    @property
    def deployment_parents(self):
        parents = []
        for label in self.labels:
            if label.key == 'csys-obj-parent' and label.value:
                parents.append(label.value)
        return parents

    def make_create_environment_execution(self, inputs=None, **params):
        if inputs is not None and self.blueprint and self.blueprint.plan:
            self._validate_inputs(inputs)
        self.create_execution = Execution(
            workflow_id='create_deployment_environment',
            deployment=self,
            status=ExecutionState.PENDING,
            parameters={'inputs': inputs, **params},
        )
        self.latest_execution = self.create_execution
        return self.create_execution

    def _validate_inputs(self, inputs):
        blueprint_inputs = self.blueprint.plan.get('inputs', {})
        allowed_inputs = set(blueprint_inputs)
        required_inputs = {
            name for name, input_spec in blueprint_inputs.items()
            if 'default' not in input_spec
        }
        provided_inputs = set(inputs)
        missing_inputs = required_inputs - provided_inputs
        undeclared_inputs = provided_inputs - allowed_inputs
        if missing_inputs:
            raise manager_exceptions.MissingRequiredDeploymentInputError(
                f'missing inputs: { ", ".join(missing_inputs) }'
            )
        if undeclared_inputs:
            raise manager_exceptions.UnknownDeploymentInputError(
                f'Cannot create deployment - unknown inputs: '
                f'{ ", ".join(undeclared_inputs) }'
            )

    def make_delete_environment_execution(self, delete_logs=True, force=False):
        return Execution(
            workflow_id='delete_deployment_environment',
            deployment=self,
            status=ExecutionState.PENDING,
            parameters={'delete_logs': delete_logs, 'force': force},
        )

    @hybrid_property
    def environment_type(self):
        for label in self.labels:
            if label.key == 'csys-env-type':
                return label.value
        return ''

    @environment_type.expression
    def environment_type(cls):
        labels_table = aliased(DeploymentLabel.__table__)
        env_type_stmt = (
            select([labels_table.c.value]).
            where(db.and_(
                    labels_table.c.key == 'csys-env-type',
                    labels_table.c._labeled_model_fk == cls._storage_id)).
            distinct().
            limit(1)
        )

        return case([(exists(env_type_stmt),
                      env_type_stmt.label('environment_type'))],
                    else_='')

    @hybrid_property
    def has_sub_deployments(self):
        return self.sub_services_count or self.sub_environments_count

    @has_sub_deployments.expression
    def has_sub_deployments(self):
        return (self.sub_services_count + self.sub_environments_count) > 0

    def get_dependencies(self, fetch_deployments=True, locking=False):
        """Dependency deployments of this deployment.

        Those are dependencies as defined by InterDeploymentDependencies.

        :param fetch_deployments: if set (the default), return deployments;
            otherwise return the InterDeploymentDependency objects
        :param locking: select using a `WITH FOR UPDATE`
        """
        return InterDeploymentDependencies.get_dependencies(
            deployment_ids=[self._storage_id],
            dependents=False,
            fetch_deployments=fetch_deployments,
            locking=locking,
        )

    def get_dependents(self, fetch_deployments=True, locking=False):
        """Dependent deployments of this deployment.

        Those are dependents as defined by InterDeploymentDependencies.
        See get_dependencies for the explanation of parameters.
        """
        return InterDeploymentDependencies.get_dependencies(
            deployment_ids=[self._storage_id],
            dependents=True,
            fetch_deployments=fetch_deployments,
            locking=locking,
        )

    def get_ancestors(self, fetch_deployments=True, locking=False):
        """Ancestor deployments of this deployment.

        Those are ancestors as defined by DeploymentLabelsDependencies.
        See get_dependencies for the explanation of parameters.
        """
        return DeploymentLabelsDependencies.get_dependencies(
            deployment_ids=[self._storage_id],
            dependents=False,
            fetch_deployments=fetch_deployments,
            locking=locking,
        )

    def get_descendants(self, fetch_deployments=True, locking=False):
        """Descendant deployments of this deployment.

        Those are descendants as defined by DeploymentLabelsDependencies.
        See get_dependencies for the explanation of parameters.
        """
        return DeploymentLabelsDependencies.get_dependencies(
            deployment_ids=[self._storage_id],
            dependents=True,
            fetch_deployments=fetch_deployments,
            locking=locking,
        )

    def get_all_dependencies(self, *args, **kwargs):
        """Both dependencies, and ancestors, of this deployment"""
        return set(
            self.get_ancestors(*args, **kwargs) +
            self.get_dependencies(*args, **kwargs)
        )

    def get_all_dependents(self, *args, **kwargs):
        """Both dependents, and descendants, of this deployment"""
        return set(
            self.get_dependents(*args, **kwargs) +
            self.get_descendants(*args, **kwargs)
        )

    def check_unique_query(self):
        if self.guaranteed_unique:
            return
        return super(Deployment, self).check_unique_query()


class DeploymentGroup(CreatedAtMixin, SQLResourceBase):
    __tablename__ = 'deployment_groups'
    description = db.Column(db.Text)
    default_inputs = db.Column(JSONString)
    creation_counter = db.Column(db.Integer, default=0, nullable=False)
    _default_blueprint_fk = foreign_key(
        Blueprint._storage_id,
        ondelete='SET NULL',
        nullable=True)

    deployment_ids = association_proxy('deployments', 'id')
    default_blueprint_id = association_proxy('default_blueprint', 'id')

    @declared_attr
    def default_blueprint(cls):
        return one_to_many_relationship(
            cls, Blueprint, cls._default_blueprint_fk)

    @declared_attr
    def deployments(cls):
        return many_to_many_relationship(cls, Deployment, unique=True)

    @classproperty
    def response_fields(cls):
        fields = super(DeploymentGroup, cls).response_fields
        fields['deployment_ids'] = flask_fields.List(
            flask_fields.String()
        )
        fields['labels'] = flask_fields.List(
            flask_fields.Nested(Label.resource_fields))
        fields['default_blueprint_id'] = flask_fields.String()
        return fields

    def to_response(self, include=None, get_data=False, **kwargs):
        include = include or self.response_fields
        response = super(DeploymentGroup, self).to_response(include, **kwargs)
        if get_data or 'labels' in include:
            response['labels'] = self.list_labels(self.labels)
        return response


class LabelBase(CreatedAtMixin, SQLModelBase):
    """An abstract class for the different labels models."""
    __abstract__ = True

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    key = db.Column(db.Text, nullable=False, index=True)
    value = db.Column(db.Text, nullable=False, index=True)

    labeled_model = None

    @declared_attr
    def _creator_id(cls):
        return foreign_key(User.id)

    @declared_attr
    def creator(cls):
        return one_to_many_relationship(cls, User, cls._creator_id, 'id')


class DeploymentLabel(LabelBase):
    __tablename__ = 'deployments_labels'
    __table_args__ = (
        db.UniqueConstraint(
            'key', 'value', '_labeled_model_fk'),
    )
    labeled_model = Deployment

    _labeled_model_fk = foreign_key(Deployment._storage_id)

    @declared_attr
    def deployment(cls):
        return db.relationship(
            Deployment, lazy='joined',
            backref=db.backref('labels', cascade='all, delete-orphan'))

    visibility = association_proxy('deployment', 'visibility')
    _tenant_id = association_proxy('deployment', '_tenant_id')


class BlueprintLabel(LabelBase):
    __tablename__ = 'blueprints_labels'
    __table_args__ = (
        db.UniqueConstraint(
            'key', 'value', '_labeled_model_fk'),
    )
    labeled_model = Blueprint

    _labeled_model_fk = foreign_key(Blueprint._storage_id)

    @declared_attr
    def blueprint(cls):
        return db.relationship(
            Blueprint, lazy='joined',
            backref=db.backref('labels', cascade='all, delete-orphan'))

    visibility = association_proxy('blueprint', 'visibility')
    _tenant_id = association_proxy('blueprint', '_tenant_id')


class DeploymentGroupLabel(LabelBase):
    __tablename__ = 'deployment_groups_labels'
    __table_args__ = (
        db.UniqueConstraint(
            'key', 'value', '_labeled_model_fk'),
    )
    labeled_model = DeploymentGroup

    _labeled_model_fk = foreign_key(DeploymentGroup._storage_id)

    @declared_attr
    def deployment_group(cls):
        return db.relationship(
            DeploymentGroup, lazy='joined',
            backref=db.backref('labels', cascade='all, delete-orphan'))

    visibility = association_proxy('deployment_group', 'visibility')
    _tenant_id = association_proxy('deployment_group', '_tenant_id')


class FilterBase(CreatedAtMixin, SQLResourceBase):
    __abstract__ = True
    _extra_fields = {'labels_filter_rules': flask_fields.Raw,
                     'attrs_filter_rules': flask_fields.Raw}

    value = db.Column(JSONString, nullable=True)
    updated_at = db.Column(UTCDateTime)
    is_system_filter = db.Column(db.Boolean,
                                 nullable=False,
                                 index=True,
                                 default=False)

    @property
    def labels_filter_rules(self):
        return [filter_rule for filter_rule in self.value
                if filter_rule['type'] == 'label']

    @property
    def attrs_filter_rules(self):
        return [filter_rule for filter_rule in self.value
                if filter_rule['type'] == 'attribute']


class DeploymentsFilter(FilterBase):
    __tablename__ = 'deployments_filters'
    __table_args__ = (
        db.Index(
            'deployments_filters_id__tenant_id_idx',
            'id', '_tenant_id',
            unique=True
        ),
    )


class BlueprintsFilter(FilterBase):
    __tablename__ = 'blueprints_filters'
    __table_args__ = (
        db.Index(
            'blueprints_filters_id__tenant_id_idx',
            'id', '_tenant_id',
            unique=True
        ),
    )


class Execution(CreatedAtMixin, SQLResourceBase):
    def __init__(self, **kwargs):
        # allow-custom must be set before other attributes, necessarily
        # before parameters
        self.allow_custom_parameters = kwargs.pop(
            'allow_custom_parameters', False)
        super().__init__(**kwargs)

    __tablename__ = 'executions'
    STATUS_DISPLAY_NAMES = {
        ExecutionState.TERMINATED: 'completed'
    }
    _extra_fields = {
        'status_display': flask_fields.String
    }
    __table_args__ = (
        db.Index(
            'executions_dep_fk_isw_vis_tenant_id_idx',
            '_deployment_fk', 'is_system_workflow', 'visibility', '_tenant_id'
        ),
    )
    # Can we skip check_unique because the exec. is part of dep. group creation
    guaranteed_unique = False

    id = db.Column(db.Text, index=True, default=lambda: str(uuid.uuid4()))
    ended_at = db.Column(UTCDateTime, nullable=True, index=True)
    error = db.Column(db.Text)
    is_system_workflow = db.Column(db.Boolean, nullable=False, index=True,
                                   default=False)
    parameters = db.Column(db.PickleType(protocol=2))
    status = db.Column(
        db.Enum(*ExecutionState.STATES, name='execution_status')
    )
    workflow_id = db.Column(db.Text, nullable=False)
    started_at = db.Column(UTCDateTime, nullable=True)
    scheduled_for = db.Column(UTCDateTime, nullable=True)
    is_dry_run = db.Column(db.Boolean, nullable=False, default=False)
    token = db.Column(db.String(100), nullable=True, index=True)
    resume = db.Column(db.Boolean, nullable=False, server_default='false')

    _deployment_fk = foreign_key(Deployment._storage_id, nullable=True)

    total_operations = db.Column(db.Integer, nullable=True)
    finished_operations = db.Column(db.Integer, nullable=True)
    allow_custom_parameters = db.Column(db.Boolean, nullable=False,
                                        server_default='false')
    execution_group_id = association_proxy('execution_groups', 'id')
    deployment_display_name = association_proxy('deployment', 'display_name')

    @classproperty
    def autoload_relationships(cls):
        return [
            cls.deployment,
        ]

    def __repr__(self):
        return (
            f'<Execution id=`{self.id}` tenant=`{self.tenant_name}` '
            f'status=`{self.status}` workflow_id=`{self.workflow_id}`>'
        )

    @declared_attr
    def deployment(cls):
        return one_to_many_relationship(
            cls, Deployment, cls._deployment_fk,
            backref=db.backref(
                'executions', passive_deletes=True, cascade='all'))

    deployment_id = association_proxy('deployment', 'id')
    blueprint_id = db.Column(db.Text, nullable=True)

    @hybrid_property
    def status_display(self):
        status = self.status
        return self.STATUS_DISPLAY_NAMES.get(status, status)

    @status_display.expression
    def status_display(cls):
        table = cls.__table__
        cases = [
            (table.c.status == status, label)
            for status, label in cls.STATUS_DISPLAY_NAMES.items()
        ]
        return case(cases, else_=db.cast(table.c.status, db.Text))

    def _get_identifier_dict(self):
        id_dict = super(Execution, self)._get_identifier_dict()
        id_dict['status'] = self.status
        return id_dict

    @classproperty
    def resource_fields(cls):
        fields = super(Execution, cls).resource_fields
        fields.pop('token')
        fields['deployment_display_name'] = flask_fields.String()
        return fields

    @classproperty
    def response_fields(cls):
        if not hasattr(cls, '_cached_execution_fields'):
            fields = super(Execution, cls).response_fields
            fields.pop('execution_group_id')
            cls._cached_execution_fields = fields
        return cls._cached_execution_fields

    @validates('deployment')
    def _set_deployment(self, key, deployment):
        self._set_parent(deployment)
        if self.blueprint_id is None:
            self.blueprint_id = deployment.blueprint_id
        if self.parameters is not None and self.workflow_id is not None:
            self.merge_workflow_parameters(
                self.parameters, deployment, self.workflow_id)
        return deployment

    @validates('workflow_id')
    def _validate_workflow_id(self, key, workflow_id):
        if self.parameters is not None and self.deployment is not None:
            self.merge_workflow_parameters(
                self.parameters, self.deployment, workflow_id)
        return workflow_id

    @validates('parameters')
    def _validate_parameters(self, key, parameters):
        parameters = parameters or {}
        if self.workflow_id is not None and self.deployment is not None:
            self.merge_workflow_parameters(
                parameters, self.deployment, self.workflow_id)
        return parameters

    def _system_workflow_task_name(self, wf_id):
        return {
            'create_snapshot': 'cloudify_system_workflows.snapshot.create',
            'restore_snapshot': 'cloudify_system_workflows.snapshot.restore',
            'uninstall_plugin': 'cloudify_system_workflows.plugins.uninstall',
            'create_deployment_environment':
                'cloudify_system_workflows.deployment_environment.create',
            'delete_deployment_environment':
                'cloudify_system_workflows.deployment_environment.delete',
            'update_plugin': 'cloudify_system_workflows.plugins.update',
            'upload_blueprint': 'cloudify_system_workflows.blueprint.upload',
            'csys_update_deployment':
                'cloudify_system_workflows.deployment_environment.'
                'update_deployment',
            'update':
                'cloudify_system_workflows.deployment_update.workflow.'
                'update_deployment',
        }.get(wf_id)

    def get_workflow(self, deployment=None, workflow_id=None):
        deployment = deployment or self.deployment
        workflow_id = workflow_id or self.workflow_id

        if deployment and deployment.workflows\
                and workflow_id in deployment.workflows:
            return deployment.workflows[workflow_id]
        system_task_name = self._system_workflow_task_name(workflow_id)
        if system_task_name:
            self.allow_custom_parameters = True
            return {'operation': system_task_name}
        if deployment:
            raise manager_exceptions.NonexistentWorkflowError(
                f'Workflow {workflow_id} does not exist in the '
                f'deployment {deployment.id}')
        else:
            raise manager_exceptions.NonexistentWorkflowError(
                f'Builtin workflow {workflow_id} does not exist')

    def merge_workflow_parameters(self, parameters, deployment, workflow_id):
        if not deployment or not deployment.workflows:
            return

        # Keep this import line here because of circular dependencies
        from manager_rest.rest.search_utils \
            import GetValuesWithStorageManager

        workflow = self.get_workflow(deployment, workflow_id)
        workflow_parameters = workflow.get('parameters', {})
        custom_parameters = parameters.keys() - workflow_parameters.keys()
        if not self.allow_custom_parameters and custom_parameters:
            raise manager_exceptions.IllegalExecutionParametersError(
                f'Workflow "{workflow_id}" does not have the following '
                f'parameters declared: { ",".join(custom_parameters) }. '
                f'Remove these parameters or use the flag for allowing '
                f'custom parameters') from None

        wrong_types = {}
        for name, param in workflow_parameters.items():
            declared_type = param.get('type')
            if declared_type is None or name not in parameters:
                continue
            try:
                parameters[name] = self._convert_param_type(
                        parameters[name], declared_type)
            except ValueError:
                wrong_types[name] = declared_type
        if wrong_types:
            raise manager_exceptions.IllegalExecutionParametersError('\n'.join(
                f'Parameter "{n}" must be of type {t}'
                for n, t in wrong_types.items())
            )

        for name, param in workflow_parameters.items():
            if 'default' in param:
                parameters.setdefault(name, param['default'])

        constraint_violations = {}
        for name, param in workflow_parameters.items():
            constraints = extract_constraints(param)
            if name not in parameters:
                continue
            try:
                getter = GetValuesWithStorageManager(get_storage_manager(),
                                                     self.deployment_id)
                if param.get('type') not in ['blueprint_id', 'deployment_id',
                                             'capability_value', 'secret_key',
                                             'node_id'] \
                        and not constraints:
                    continue
                validate_input_value(name, constraints, parameters[name],
                                     param.get('type'), getter)
            except (dsl_exceptions.DSLParsingException,
                    dsl_exceptions.ConstraintException) as ex:
                constraint_violations[name] = ex
        if constraint_violations:
            raise manager_exceptions.IllegalExecutionParametersError('\n'.join(
                f'Parameter "{n}" does not meet its constraints: {e}'
                for n, e in constraint_violations.items()
            ))

        missing_parameters = workflow_parameters.keys() - parameters.keys()
        if missing_parameters:
            raise manager_exceptions.IllegalExecutionParametersError(
                f'Workflow "{workflow_id}" must be provided with the following'
                f' parameters to execute: { ",".join(missing_parameters) }'
            ) from None

    def _convert_param_type(self, param, target_type):
        if target_type == 'integer':
            return int(param)
        elif target_type == 'boolean':<commit_msg>RD-4711 Always validate for default deployment_id (#3645)<commit_after>import hashlib
import typing
import uuid

from os import path
from datetime import datetime
from collections import namedtuple

from flask_restful import fields as flask_fields

from sqlalchemy import case
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy import func, select, table, column, exists
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.orm import validates, aliased
from sqlalchemy.sql.schema import CheckConstraint

from cloudify.constants import MGMTWORKER_QUEUE
from cloudify.models_states import (AgentState,
                                    SnapshotState,
                                    ExecutionState,
                                    VisibilityState,
                                    DeploymentModificationState,
                                    DeploymentState)
from cloudify.cryptography_utils import decrypt
from dsl_parser.constants import (WORKFLOW_PLUGINS_TO_INSTALL,
                                  TYPES_BASED_ON_DB_ENTITIES)
from dsl_parser.constraints import extract_constraints, validate_input_value
from dsl_parser import exceptions as dsl_exceptions

from manager_rest import config, manager_exceptions
from manager_rest.rest.responses import Workflow, Label
from manager_rest.utils import (get_rrule,
                                classproperty,
                                files_in_folder)
from manager_rest.deployment_update.constants import ACTION_TYPES, ENTITY_TYPES
from manager_rest.constants import (FILE_SERVER_PLUGINS_FOLDER,
                                    FILE_SERVER_RESOURCES_FOLDER,
                                    AUDIT_OPERATIONS)

from .models_base import (
    CreatedAtMixin,
    db,
    JSONString,
    UTCDateTime,
)
from .management_models import User, Manager
from .resource_models_base import SQLResourceBase, SQLModelBase
from .relationships import (
    foreign_key,
    one_to_many_relationship,
    many_to_many_relationship
)
from manager_rest.storage.storage_manager import get_storage_manager


if typing.TYPE_CHECKING:
    from manager_rest.resource_manager import ResourceManager
    from manager_rest.storage.storage_manager import SQLStorageManager


RELATIONSHIP = 'relationship'
NODE = 'node'


# region Top Level Resources

class Blueprint(CreatedAtMixin, SQLResourceBase):
    __tablename__ = 'blueprints'
    __table_args__ = (
        db.Index(
            'blueprints_id__tenant_id_idx',
            'id', '_tenant_id',
            unique=True
        ),
    )

    skipped_fields = dict(
        SQLResourceBase.skipped_fields,
        v1=['main_file_name', 'description']
    )

    main_file_name = db.Column(db.Text)
    plan = db.Column(db.PickleType(protocol=2))
    updated_at = db.Column(UTCDateTime)
    description = db.Column(db.Text)
    is_hidden = db.Column(db.Boolean, nullable=False, default=False)
    state = db.Column(db.Text)
    error = db.Column(db.Text)
    error_traceback = db.Column(db.Text)
    _upload_execution_fk = foreign_key('executions._storage_id',
                                       nullable=True,
                                       ondelete='SET NULL',
                                       deferrable=True,
                                       initially='DEFERRED',
                                       use_alter=True)

    @classproperty
    def labels_model(cls):
        return BlueprintLabel

    @declared_attr
    def labels(cls):
        # labels are defined as `backref` in DeploymentsLabel model
        return None

    @declared_attr
    def upload_execution(cls):
        return db.relationship('Execution',
                               foreign_keys=[cls._upload_execution_fk],
                               cascade='all, delete')

    @classproperty
    def response_fields(cls):
        fields = super(Blueprint, cls).response_fields
        fields['labels'] = flask_fields.List(
            flask_fields.Nested(Label.resource_fields))
        fields['upload_execution'] = flask_fields.Nested(
            Execution.resource_fields)
        return fields

    @classproperty
    def allowed_filter_attrs(cls):
        return ['created_by', 'state', 'id', 'tenant_name']

    def to_response(self, include=None, **kwargs):
        include = include or self.response_fields
        blueprint_dict = super(Blueprint, self).to_response(include, **kwargs)
        if 'labels' in include:
            blueprint_dict['labels'] = self.list_labels(self.labels)
        return blueprint_dict


class Snapshot(CreatedAtMixin, SQLResourceBase):
    __tablename__ = 'snapshots'
    __table_args__ = (
        db.Index(
            'snapshots_id__tenant_id_idx',
            'id', '_tenant_id',
            unique=True
        ),
    )

    status = db.Column(db.Enum(*SnapshotState.STATES, name='snapshot_status'))
    error = db.Column(db.Text)


class Plugin(SQLResourceBase):
    __tablename__ = 'plugins'
    __table_args__ = (
        db.Index(
            'plugins_name_version__tenant_id_idx',
            'package_name', 'package_version', '_tenant_id', 'distribution',
            'distribution_release', 'distribution_version',
            unique=True
        ),
    )
    _extra_fields = {'installation_state': flask_fields.Raw}

    archive_name = db.Column(db.Text, nullable=False, index=True)
    distribution = db.Column(db.Text)
    distribution_release = db.Column(db.Text)
    distribution_version = db.Column(db.Text)
    excluded_wheels = db.Column(db.PickleType(protocol=2))
    package_name = db.Column(db.Text, nullable=False, index=True)
    package_source = db.Column(db.Text)
    package_version = db.Column(db.Text)
    supported_platform = db.Column(db.PickleType(protocol=2))
    supported_py_versions = db.Column(db.PickleType(protocol=2))
    uploaded_at = db.Column(UTCDateTime, nullable=False, index=True)
    wheels = db.Column(db.PickleType(protocol=2), nullable=False)
    title = db.Column(db.Text)
    blueprint_labels = db.Column(JSONString)
    labels = db.Column(JSONString)
    resource_tags = db.Column(JSONString)

    def yaml_file_path(self):
        plugin_dir = path.join(config.instance.file_server_root,
                               FILE_SERVER_PLUGINS_FOLDER,
                               self.id)
        if not path.isdir(plugin_dir):
            return None
        yaml_files = files_in_folder(plugin_dir, '*.yaml')
        return yaml_files[0] if yaml_files else None

    def _yaml_file_name(self):
        yaml_path = self.yaml_file_path()
        return path.basename(yaml_path) if yaml_path else ''

    @property
    def file_server_path(self):
        file_name = self._yaml_file_name()
        if not file_name:
            return ''
        return path.join(FILE_SERVER_RESOURCES_FOLDER,
                         FILE_SERVER_PLUGINS_FOLDER,
                         self.id,
                         file_name)

    @property
    def yaml_url_path(self):
        if not self._yaml_file_name():
            return ''
        return 'plugin:{0}?version={1}&distribution={2}'.format(
            self.package_name,
            self.package_version,
            self.distribution
        )

    @classproperty
    def response_fields(cls):
        fields = super(Plugin, cls).response_fields
        fields['file_server_path'] = flask_fields.String
        fields['yaml_url_path'] = flask_fields.String
        return fields

    def to_response(self, include=None, get_data=False, **kwargs):
        include = include or self.response_fields
        plugin_dict = super(Plugin, self).to_response(include, **kwargs)
        if not get_data:
            plugin_dict['file_server_path'] = ''
        if 'installation_state' in plugin_dict \
                and 'installation_state' in include:
            plugin_dict['installation_state'] = [
                s.to_dict() for s in plugin_dict['installation_state']]
        return plugin_dict

    @declared_attr
    def installation_state(cls):
        return db.relationship('_PluginState', cascade='delete',
                               passive_deletes=True)

    def check_unique_query(self):
        query = self.__class__.query \
            .filter(self.__class__.package_name == self.package_name) \
            .filter(self.__class__.archive_name == self.archive_name)
        if self.visibility != VisibilityState.GLOBAL:
            tenant_or_global_filter = db.or_(
                self.__class__.tenant == self.tenant,
                self.__class__.visibility == VisibilityState.GLOBAL
            )
            query = query.filter(tenant_or_global_filter)
        return query


class _PluginState(SQLModelBase):
    __tablename__ = 'plugins_states'
    _extra_fields = {'manager': flask_fields.Raw}

    _storage_id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    _plugin_fk = foreign_key(Plugin._storage_id)
    _manager_fk = foreign_key('managers.id', nullable=True)
    _agent_fk = foreign_key('agents._storage_id', nullable=True)
    state = db.Column(db.Text)
    error = db.Column(db.Text)

    @classmethod
    def unique_id(cls):
        return '_storage_id'

    def __repr__(self):
        return '<PluginState plugin={0} state={1}>'.format(
            self._plugin_fk, self.state)

    def to_dict(self, suppress_error=False):
        rv = super(_PluginState, self).to_dict(suppress_error)
        rv['manager'] = self.manager.hostname if self.manager else None
        rv['agent'] = self.agent.name if self.agent else None
        return rv

    @declared_attr
    def manager(cls):
        return db.relationship('Manager', lazy='joined')

    @declared_attr
    def agent(cls):
        return db.relationship('Agent', lazy='joined')


class Secret(CreatedAtMixin, SQLResourceBase):
    __tablename__ = 'secrets'
    __table_args__ = (
        db.Index(
            'secrets_id_tenant_id_idx',
            'id', '_tenant_id',
            unique=True
        ),
    )

    value = db.Column(db.Text)
    updated_at = db.Column(UTCDateTime)
    is_hidden_value = db.Column(db.Boolean, nullable=False, default=False)

    @hybrid_property
    def key(self):
        return self.id

    @classproperty
    def resource_fields(cls):
        fields = super(Secret, cls).resource_fields
        fields['key'] = fields.pop('id')
        return fields

    @classproperty
    def allowed_filter_attrs(cls):
        return ['key']


class Site(CreatedAtMixin, SQLResourceBase):
    __tablename__ = 'sites'
    __table_args__ = (
        db.Index(
            'site_name__tenant_id_idx',
            'name', '_tenant_id',
            unique=True
        ),
    )

    name = db.Column(db.Text, nullable=False)
    latitude = db.Column(db.Float, nullable=True)
    longitude = db.Column(db.Float, nullable=True)

    _extra_fields = {
        'location': flask_fields.String
    }

    @property
    def location(self):
        if (self.latitude is None) or (self.longitude is None):
            return None
        return "{0}, {1}".format(self.latitude, self.longitude)

    @classproperty
    def response_fields(cls):
        if not hasattr(cls, '_cached_site_fields'):
            fields = super(Site, cls).response_fields
            fields.pop('id')
            cls._cached_site_fields = fields
        return cls._cached_site_fields

# endregion

# region Derived Resources


class Deployment(CreatedAtMixin, SQLResourceBase):
    __tablename__ = 'deployments'
    __table_args__ = (
        db.Index(
            'deployments__site_fk_visibility_idx',
            '_blueprint_fk', '_site_fk', 'visibility', '_tenant_id'
        ),
        db.Index(
            'deployments_id__tenant_id_idx',
            'id', '_tenant_id',
            unique=True
        ),
        db.Index(
            'deployments__latest_execution_fk_idx',
            '_latest_execution_fk',
            unique=True
        ),
    )
    _extra_fields = {
        'workdir_zip': flask_fields.String,
    }
    skipped_fields = dict(
        SQLResourceBase.skipped_fields,
        v1=['scaling_groups'],
        v2=['scaling_groups']
    )

    # Can we skip check_unique because it was checked in group dep. creation
    guaranteed_unique = False

    # This will be overridden when a workdir is being retrieved
    workdir_zip = None

    description = db.Column(db.Text)
    inputs = db.Column(db.PickleType(protocol=2))
    groups = db.Column(db.PickleType(protocol=2))
    permalink = db.Column(db.Text)
    policy_triggers = db.Column(db.PickleType(protocol=2))
    policy_types = db.Column(db.PickleType(protocol=2))
    outputs = db.Column(db.PickleType(protocol=2, comparator=lambda *a: False))
    capabilities = db.Column(db.PickleType(
        protocol=2, comparator=lambda *a: False))
    scaling_groups = db.Column(db.PickleType(protocol=2))
    updated_at = db.Column(UTCDateTime)
    workflows = db.Column(db.PickleType(
        protocol=2, comparator=lambda *a: False))
    runtime_only_evaluation = db.Column(db.Boolean, default=False)
    installation_status = db.Column(db.Enum(
        DeploymentState.ACTIVE,
        DeploymentState.INACTIVE,
        name="installation_status"
    ))
    deployment_status = db.Column(db.Enum(
            DeploymentState.GOOD,
            DeploymentState.IN_PROGRESS,
            DeploymentState.REQUIRE_ATTENTION,
            name='deployment_status'
        )
    )
    sub_services_status = db.Column(db.Enum(
            DeploymentState.GOOD,
            DeploymentState.IN_PROGRESS,
            DeploymentState.REQUIRE_ATTENTION,
            name='deployment_status'
        ))
    sub_environments_status = db.Column(db.Enum(
            DeploymentState.GOOD,
            DeploymentState.IN_PROGRESS,
            DeploymentState.REQUIRE_ATTENTION,
            name='deployment_status'
        ))
    sub_services_count = db.Column(
        db.Integer, nullable=False, server_default='0')
    sub_environments_count = db.Column(
        db.Integer, nullable=False, server_default='0')
    display_name = db.Column(
        db.Text, nullable=False, index=True,
        default=lambda ctx: ctx.get_current_parameters().get('id')
    )
    resource_tags = db.Column(JSONString)
    _blueprint_fk = foreign_key(Blueprint._storage_id)
    _site_fk = foreign_key(Site._storage_id,
                           nullable=True,
                           ondelete='SET NULL')
    _create_execution_fk = foreign_key('executions._storage_id',
                                       nullable=True,
                                       ondelete='SET NULL',
                                       deferrable=True,
                                       initially='DEFERRED',
                                       use_alter=True)

    _latest_execution_fk = foreign_key('executions._storage_id',
                                       nullable=True,
                                       ondelete='SET NULL',
                                       deferrable=True,
                                       initially='DEFERRED',
                                       use_alter=True,
                                       unique=True)

    deployment_group_id = association_proxy('deployment_groups', 'id')
    latest_execution_finished_operations = association_proxy(
        'latest_execution', 'finished_operations')
    latest_execution_total_operations = association_proxy(
        'latest_execution', 'total_operations')

    @classproperty
    def autoload_relationships(cls):
        return [
            cls.create_execution,
            cls.latest_execution,
            cls.deployment_groups,
            cls.labels,
            cls.schedules,
        ]

    @classproperty
    def labels_model(cls):
        return DeploymentLabel

    @declared_attr
    def create_execution(cls):
        """The create-deployment-environment execution for this deployment"""
        return db.relationship('Execution',
                               foreign_keys=[cls._create_execution_fk],
                               cascade='all, delete',
                               post_update=True)

    @declared_attr
    def latest_execution(cls):
        return db.relationship('Execution',
                               foreign_keys=[cls._latest_execution_fk],
                               cascade='all, delete',
                               post_update=True)

    @declared_attr
    def blueprint(cls):
        return one_to_many_relationship(cls, Blueprint, cls._blueprint_fk)

    blueprint_id = association_proxy('blueprint', 'id')

    @declared_attr
    def site(cls):
        return one_to_many_relationship(cls, Site, cls._site_fk, cascade=False)

    site_name = association_proxy('site', 'name')
    latest_execution_status = association_proxy('latest_execution', 'status')

    @classproperty
    def response_fields(cls):
        if not hasattr(cls, '_cached_deployment_fields'):
            fields = super(Deployment, cls).response_fields
            fields['labels'] = flask_fields.List(
                flask_fields.Nested(Label.resource_fields))
            fields.pop('deployment_group_id', None)
            fields['workflows'] = flask_fields.List(
                flask_fields.Nested(Workflow.resource_fields)
            )
            fields['schedules'] = flask_fields.List(
                flask_fields.Nested(ExecutionSchedule.resource_fields))
            fields['deployment_groups'] = \
                flask_fields.List(flask_fields.String)
            fields['latest_execution_status'] = flask_fields.String()
            fields['latest_execution_total_operations'] = \
                flask_fields.Integer()
            fields['latest_execution_finished_operations'] = \
                flask_fields.Integer()
            fields['has_sub_deployments'] = flask_fields.Boolean()
            fields['create_execution'] = flask_fields.String()
            fields['latest_execution'] = flask_fields.String()
            cls._cached_deployment_fields = fields
        return cls._cached_deployment_fields

    @classproperty
    def allowed_filter_attrs(cls):
        return ['blueprint_id', 'created_by', 'site_name', 'schedules',
                'tenant_name', 'display_name', 'installation_status']

    def to_response(self, include=None, **kwargs):
        include = include or self.response_fields
        dep_dict = super(Deployment, self).to_response(
            include=include, **kwargs)
        if 'workflows' in include:
            dep_dict['workflows'] = self._list_workflows(self.workflows)
        if 'labels' in include:
            dep_dict['labels'] = self.list_labels(self.labels)
        if 'deployment_groups' in include:
            dep_dict['deployment_groups'] = \
                [g.id for g in self.deployment_groups]
        if 'latest_execution_status' in include:
            dep_dict['latest_execution_status'] = \
                DeploymentState.EXECUTION_STATES_SUMMARY.get(
                    self.latest_execution_status)
        if 'installation_status' in include:
            if not dep_dict.get('installation_status'):
                dep_dict['installation_status'] = DeploymentState.INACTIVE
        if 'latest_execution_total_operations' in include:
            dep_dict['latest_execution_total_operations'] = \
                self.latest_execution_total_operations
        if 'latest_execution_finished_operations' in include:
            dep_dict['latest_execution_finished_operations'] = \
                self.latest_execution_finished_operations
        if 'create_execution' in include:
            dep_dict['create_execution'] = \
                self.create_execution.id if self.create_execution else None
        if 'latest_execution' in include:
            dep_dict['latest_execution'] = \
                self.latest_execution.id if self.latest_execution else None
        return dep_dict

    @staticmethod
    def _list_workflows(deployment_workflows):
        if deployment_workflows is None:
            return None

        return [Workflow(name=wf_name,
                         created_at=None,
                         plugin=wf.get('plugin', ''),
                         operation=wf.get('operation', ''),
                         parameters=wf.get('parameters', dict()),
                         is_cascading=wf.get('is_cascading', False))
                for wf_name, wf in deployment_workflows.items()]

    @classmethod
    def compare_statuses(
                cls, *statuses: typing.Optional[str]
            ) -> typing.Optional[str]:
        """Unify multiple DeploymentStates into a single state.

        Choose the "worst" possible outcome based on the given states,
        ie. if there's one that requires attention, then the overall status
        is also "requires attention".
        """
        if not statuses:
            return None
        importance = {
            DeploymentState.GOOD: 1,
            DeploymentState.IN_PROGRESS: 2,
            DeploymentState.REQUIRE_ATTENTION: 3
        }
        return max(statuses, key=lambda st: importance.get(st, 0))

    def evaluate_sub_deployments_statuses(self):
        """
        Evaluate the deployment statuses per deployment using the following
        three statuses
        1. sub_environments_status
        2. sub_services_status
        3. deployment_status
        This is useful and prerequisite before propagate the source
        deployments statuses to the target
        :return: Tuple of end result of `sub_environments_status`
         & `sub_services_status`
        :rtype Tuple
        """
        _sub_environments_status = self.sub_environments_status
        _sub_services_status = self.sub_services_status
        if self.is_environment:
            _sub_environments_status = \
                self.compare_statuses(
                    self.sub_environments_status,
                    self.deployment_status
                )

        else:
            _sub_services_status = \
                self.compare_statuses(
                    self.sub_services_status,
                    self.deployment_status
                )
        return _sub_services_status, _sub_environments_status

    def evaluate_deployment_status(self, exclude_sub_deployments=False):
        """
        Evaluate the overall deployment status based on installation status
        and latest execution object
        :return: deployment_status: Overall deployment status
        """
        deployment_status = self.decide_deployment_status(
            self.latest_execution_status,
            self.installation_status,
            self.sub_services_status,
            self.sub_environments_status
        )
        self.deployment_status = deployment_status
        return deployment_status

    @classmethod
    def decide_deployment_status(
        cls,
        latest_execution_status,
        installation_status,
        sub_services_status,
        sub_environments_status,
    ):
        latest_status = DeploymentState.EXECUTION_STATES_SUMMARY.get(
            latest_execution_status)
        if latest_status == DeploymentState.IN_PROGRESS:
            deployment_status = DeploymentState.IN_PROGRESS
        elif latest_status == DeploymentState.FAILED \
                or installation_status == DeploymentState.INACTIVE:
            deployment_status = DeploymentState.REQUIRE_ATTENTION
        else:
            deployment_status = DeploymentState.GOOD

        # Check whether or not deployment has services or environments
        # attached to it, so that we can consider that while evaluating the
        # deployment status
        if sub_services_status:
            deployment_status = cls.compare_statuses(
                sub_services_status,
                deployment_status
            )
        if sub_environments_status:
            deployment_status = cls.compare_statuses(
                sub_environments_status,
                deployment_status
            )
        return deployment_status

    @property
    def is_environment(self):
        target_key = '<KEY>'
        target_value = 'environment'
        for label in self.labels:
            if label.key == target_key and label.value == target_value:
                return True
        return False

    @property
    def deployment_parents(self):
        parents = []
        for label in self.labels:
            if label.key == 'csys-obj-parent' and label.value:
                parents.append(label.value)
        return parents

    def make_create_environment_execution(self, inputs=None, **params):
        if inputs is not None and self.blueprint and self.blueprint.plan:
            self._validate_inputs(inputs)
        self.create_execution = Execution(
            workflow_id='create_deployment_environment',
            deployment=self,
            status=ExecutionState.PENDING,
            parameters={'inputs': inputs, **params},
        )
        self.latest_execution = self.create_execution
        return self.create_execution

    def _validate_inputs(self, inputs):
        blueprint_inputs = self.blueprint.plan.get('inputs', {})
        allowed_inputs = set(blueprint_inputs)
        required_inputs = {
            name for name, input_spec in blueprint_inputs.items()
            if 'default' not in input_spec
        }
        provided_inputs = set(inputs)
        missing_inputs = required_inputs - provided_inputs
        undeclared_inputs = provided_inputs - allowed_inputs
        if missing_inputs:
            raise manager_exceptions.MissingRequiredDeploymentInputError(
                f'missing inputs: { ", ".join(missing_inputs) }'
            )
        if undeclared_inputs:
            raise manager_exceptions.UnknownDeploymentInputError(
                f'Cannot create deployment - unknown inputs: '
                f'{ ", ".join(undeclared_inputs) }'
            )

    def make_delete_environment_execution(self, delete_logs=True, force=False):
        return Execution(
            workflow_id='delete_deployment_environment',
            deployment=self,
            status=ExecutionState.PENDING,
            parameters={'delete_logs': delete_logs, 'force': force},
        )

    @hybrid_property
    def environment_type(self):
        for label in self.labels:
            if label.key == 'csys-env-type':
                return label.value
        return ''

    @environment_type.expression
    def environment_type(cls):
        labels_table = aliased(DeploymentLabel.__table__)
        env_type_stmt = (
            select([labels_table.c.value]).
            where(db.and_(
                    labels_table.c.key == 'csys-env-type',
                    labels_table.c._labeled_model_fk == cls._storage_id)).
            distinct().
            limit(1)
        )

        return case([(exists(env_type_stmt),
                      env_type_stmt.label('environment_type'))],
                    else_='')

    @hybrid_property
    def has_sub_deployments(self):
        return self.sub_services_count or self.sub_environments_count

    @has_sub_deployments.expression
    def has_sub_deployments(self):
        return (self.sub_services_count + self.sub_environments_count) > 0

    def get_dependencies(self, fetch_deployments=True, locking=False):
        """Dependency deployments of this deployment.

        Those are dependencies as defined by InterDeploymentDependencies.

        :param fetch_deployments: if set (the default), return deployments;
            otherwise return the InterDeploymentDependency objects
        :param locking: select using a `WITH FOR UPDATE`
        """
        return InterDeploymentDependencies.get_dependencies(
            deployment_ids=[self._storage_id],
            dependents=False,
            fetch_deployments=fetch_deployments,
            locking=locking,
        )

    def get_dependents(self, fetch_deployments=True, locking=False):
        """Dependent deployments of this deployment.

        Those are dependents as defined by InterDeploymentDependencies.
        See get_dependencies for the explanation of parameters.
        """
        return InterDeploymentDependencies.get_dependencies(
            deployment_ids=[self._storage_id],
            dependents=True,
            fetch_deployments=fetch_deployments,
            locking=locking,
        )

    def get_ancestors(self, fetch_deployments=True, locking=False):
        """Ancestor deployments of this deployment.

        Those are ancestors as defined by DeploymentLabelsDependencies.
        See get_dependencies for the explanation of parameters.
        """
        return DeploymentLabelsDependencies.get_dependencies(
            deployment_ids=[self._storage_id],
            dependents=False,
            fetch_deployments=fetch_deployments,
            locking=locking,
        )

    def get_descendants(self, fetch_deployments=True, locking=False):
        """Descendant deployments of this deployment.

        Those are descendants as defined by DeploymentLabelsDependencies.
        See get_dependencies for the explanation of parameters.
        """
        return DeploymentLabelsDependencies.get_dependencies(
            deployment_ids=[self._storage_id],
            dependents=True,
            fetch_deployments=fetch_deployments,
            locking=locking,
        )

    def get_all_dependencies(self, *args, **kwargs):
        """Both dependencies, and ancestors, of this deployment"""
        return set(
            self.get_ancestors(*args, **kwargs) +
            self.get_dependencies(*args, **kwargs)
        )

    def get_all_dependents(self, *args, **kwargs):
        """Both dependents, and descendants, of this deployment"""
        return set(
            self.get_dependents(*args, **kwargs) +
            self.get_descendants(*args, **kwargs)
        )

    def check_unique_query(self):
        if self.guaranteed_unique:
            return
        return super(Deployment, self).check_unique_query()


class DeploymentGroup(CreatedAtMixin, SQLResourceBase):
    __tablename__ = 'deployment_groups'
    description = db.Column(db.Text)
    default_inputs = db.Column(JSONString)
    creation_counter = db.Column(db.Integer, default=0, nullable=False)
    _default_blueprint_fk = foreign_key(
        Blueprint._storage_id,
        ondelete='SET NULL',
        nullable=True)

    deployment_ids = association_proxy('deployments', 'id')
    default_blueprint_id = association_proxy('default_blueprint', 'id')

    @declared_attr
    def default_blueprint(cls):
        return one_to_many_relationship(
            cls, Blueprint, cls._default_blueprint_fk)

    @declared_attr
    def deployments(cls):
        return many_to_many_relationship(cls, Deployment, unique=True)

    @classproperty
    def response_fields(cls):
        fields = super(DeploymentGroup, cls).response_fields
        fields['deployment_ids'] = flask_fields.List(
            flask_fields.String()
        )
        fields['labels'] = flask_fields.List(
            flask_fields.Nested(Label.resource_fields))
        fields['default_blueprint_id'] = flask_fields.String()
        return fields

    def to_response(self, include=None, get_data=False, **kwargs):
        include = include or self.response_fields
        response = super(DeploymentGroup, self).to_response(include, **kwargs)
        if get_data or 'labels' in include:
            response['labels'] = self.list_labels(self.labels)
        return response


class LabelBase(CreatedAtMixin, SQLModelBase):
    """An abstract class for the different labels models."""
    __abstract__ = True

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    key = db.Column(db.Text, nullable=False, index=True)
    value = db.Column(db.Text, nullable=False, index=True)

    labeled_model = None

    @declared_attr
    def _creator_id(cls):
        return foreign_key(User.id)

    @declared_attr
    def creator(cls):
        return one_to_many_relationship(cls, User, cls._creator_id, 'id')


class DeploymentLabel(LabelBase):
    __tablename__ = 'deployments_labels'
    __table_args__ = (
        db.UniqueConstraint(
            'key', 'value', '_labeled_model_fk'),
    )
    labeled_model = Deployment

    _labeled_model_fk = foreign_key(Deployment._storage_id)

    @declared_attr
    def deployment(cls):
        return db.relationship(
            Deployment, lazy='joined',
            backref=db.backref('labels', cascade='all, delete-orphan'))

    visibility = association_proxy('deployment', 'visibility')
    _tenant_id = association_proxy('deployment', '_tenant_id')


class BlueprintLabel(LabelBase):
    __tablename__ = 'blueprints_labels'
    __table_args__ = (
        db.UniqueConstraint(
            'key', 'value', '_labeled_model_fk'),
    )
    labeled_model = Blueprint

    _labeled_model_fk = foreign_key(Blueprint._storage_id)

    @declared_attr
    def blueprint(cls):
        return db.relationship(
            Blueprint, lazy='joined',
            backref=db.backref('labels', cascade='all, delete-orphan'))

    visibility = association_proxy('blueprint', 'visibility')
    _tenant_id = association_proxy('blueprint', '_tenant_id')


class DeploymentGroupLabel(LabelBase):
    __tablename__ = 'deployment_groups_labels'
    __table_args__ = (
        db.UniqueConstraint(
            'key', 'value', '_labeled_model_fk'),
    )
    labeled_model = DeploymentGroup

    _labeled_model_fk = foreign_key(DeploymentGroup._storage_id)

    @declared_attr
    def deployment_group(cls):
        return db.relationship(
            DeploymentGroup, lazy='joined',
            backref=db.backref('labels', cascade='all, delete-orphan'))

    visibility = association_proxy('deployment_group', 'visibility')
    _tenant_id = association_proxy('deployment_group', '_tenant_id')


class FilterBase(CreatedAtMixin, SQLResourceBase):
    __abstract__ = True
    _extra_fields = {'labels_filter_rules': flask_fields.Raw,
                     'attrs_filter_rules': flask_fields.Raw}

    value = db.Column(JSONString, nullable=True)
    updated_at = db.Column(UTCDateTime)
    is_system_filter = db.Column(db.Boolean,
                                 nullable=False,
                                 index=True,
                                 default=False)

    @property
    def labels_filter_rules(self):
        return [filter_rule for filter_rule in self.value
                if filter_rule['type'] == 'label']

    @property
    def attrs_filter_rules(self):
        return [filter_rule for filter_rule in self.value
                if filter_rule['type'] == 'attribute']


class DeploymentsFilter(FilterBase):
    __tablename__ = 'deployments_filters'
    __table_args__ = (
        db.Index(
            'deployments_filters_id__tenant_id_idx',
            'id', '_tenant_id',
            unique=True
        ),
    )


class BlueprintsFilter(FilterBase):
    __tablename__ = 'blueprints_filters'
    __table_args__ = (
        db.Index(
            'blueprints_filters_id__tenant_id_idx',
            'id', '_tenant_id',
            unique=True
        ),
    )


class Execution(CreatedAtMixin, SQLResourceBase):
    def __init__(self, **kwargs):
        # allow-custom must be set before other attributes, necessarily
        # before parameters
        self.allow_custom_parameters = kwargs.pop(
            'allow_custom_parameters', False)
        super().__init__(**kwargs)

    __tablename__ = 'executions'
    STATUS_DISPLAY_NAMES = {
        ExecutionState.TERMINATED: 'completed'
    }
    _extra_fields = {
        'status_display': flask_fields.String
    }
    __table_args__ = (
        db.Index(
            'executions_dep_fk_isw_vis_tenant_id_idx',
            '_deployment_fk', 'is_system_workflow', 'visibility', '_tenant_id'
        ),
    )
    # Can we skip check_unique because the exec. is part of dep. group creation
    guaranteed_unique = False

    id = db.Column(db.Text, index=True, default=lambda: str(uuid.uuid4()))
    ended_at = db.Column(UTCDateTime, nullable=True, index=True)
    error = db.Column(db.Text)
    is_system_workflow = db.Column(db.Boolean, nullable=False, index=True,
                                   default=False)
    parameters = db.Column(db.PickleType(protocol=2))
    status = db.Column(
        db.Enum(*ExecutionState.STATES, name='execution_status')
    )
    workflow_id = db.Column(db.Text, nullable=False)
    started_at = db.Column(UTCDateTime, nullable=True)
    scheduled_for = db.Column(UTCDateTime, nullable=True)
    is_dry_run = db.Column(db.Boolean, nullable=False, default=False)
    token = db.Column(db.String(100), nullable=True, index=True)
    resume = db.Column(db.Boolean, nullable=False, server_default='false')

    _deployment_fk = foreign_key(Deployment._storage_id, nullable=True)

    total_operations = db.Column(db.Integer, nullable=True)
    finished_operations = db.Column(db.Integer, nullable=True)
    allow_custom_parameters = db.Column(db.Boolean, nullable=False,
                                        server_default='false')
    execution_group_id = association_proxy('execution_groups', 'id')
    deployment_display_name = association_proxy('deployment', 'display_name')

    @classproperty
    def autoload_relationships(cls):
        return [
            cls.deployment,
        ]

    def __repr__(self):
        return (
            f'<Execution id=`{self.id}` tenant=`{self.tenant_name}` '
            f'status=`{self.status}` workflow_id=`{self.workflow_id}`>'
        )

    @declared_attr
    def deployment(cls):
        return one_to_many_relationship(
            cls, Deployment, cls._deployment_fk,
            backref=db.backref(
                'executions', passive_deletes=True, cascade='all'))

    deployment_id = association_proxy('deployment', 'id')
    blueprint_id = db.Column(db.Text, nullable=True)

    @hybrid_property
    def status_display(self):
        status = self.status
        return self.STATUS_DISPLAY_NAMES.get(status, status)

    @status_display.expression
    def status_display(cls):
        table = cls.__table__
        cases = [
            (table.c.status == status, label)
            for status, label in cls.STATUS_DISPLAY_NAMES.items()
        ]
        return case(cases, else_=db.cast(table.c.status, db.Text))

    def _get_identifier_dict(self):
        id_dict = super(Execution, self)._get_identifier_dict()
        id_dict['status'] = self.status
        return id_dict

    @classproperty
    def resource_fields(cls):
        fields = super(Execution, cls).resource_fields
        fields.pop('token')
        fields['deployment_display_name'] = flask_fields.String()
        return fields

    @classproperty
    def response_fields(cls):
        if not hasattr(cls, '_cached_execution_fields'):
            fields = super(Execution, cls).response_fields
            fields.pop('execution_group_id')
            cls._cached_execution_fields = fields
        return cls._cached_execution_fields

    @validates('deployment')
    def _set_deployment(self, key, deployment):
        self._set_parent(deployment)
        if self.blueprint_id is None:
            self.blueprint_id = deployment.blueprint_id
        if self.parameters is not None and self.workflow_id is not None:
            self.merge_workflow_parameters(
                self.parameters, deployment, self.workflow_id)
        return deployment

    @validates('workflow_id')
    def _validate_workflow_id(self, key, workflow_id):
        if self.parameters is not None and self.deployment is not None:
            self.merge_workflow_parameters(
                self.parameters, self.deployment, workflow_id)
        return workflow_id

    @validates('parameters')
    def _validate_parameters(self, key, parameters):
        parameters = parameters or {}
        if self.workflow_id is not None and self.deployment is not None:
            self.merge_workflow_parameters(
                parameters, self.deployment, self.workflow_id)
        return parameters

    def _system_workflow_task_name(self, wf_id):
        return {
            'create_snapshot': 'cloudify_system_workflows.snapshot.create',
            'restore_snapshot': 'cloudify_system_workflows.snapshot.restore',
            'uninstall_plugin': 'cloudify_system_workflows.plugins.uninstall',
            'create_deployment_environment':
                'cloudify_system_workflows.deployment_environment.create',
            'delete_deployment_environment':
                'cloudify_system_workflows.deployment_environment.delete',
            'update_plugin': 'cloudify_system_workflows.plugins.update',
            'upload_blueprint': 'cloudify_system_workflows.blueprint.upload',
            'csys_update_deployment':
                'cloudify_system_workflows.deployment_environment.'
                'update_deployment',
            'update':
                'cloudify_system_workflows.deployment_update.workflow.'
                'update_deployment',
        }.get(wf_id)

    def get_workflow(self, deployment=None, workflow_id=None):
        deployment = deployment or self.deployment
        workflow_id = workflow_id or self.workflow_id

        if deployment and deployment.workflows\
                and workflow_id in deployment.workflows:
            return deployment.workflows[workflow_id]
        system_task_name = self._system_workflow_task_name(workflow_id)
        if system_task_name:
            self.allow_custom_parameters = True
            return {'operation': system_task_name}
        if deployment:
            raise manager_exceptions.NonexistentWorkflowError(
                f'Workflow {workflow_id} does not exist in the '
                f'deployment {deployment.id}')
        else:
            raise manager_exceptions.NonexistentWorkflowError(
                f'Builtin workflow {workflow_id} does not exist')

    def merge_workflow_parameters(self, parameters, deployment, workflow_id):
        if not deployment or not deployment.workflows:
            return

        # Keep this import line here because of circular dependencies
        from manager_rest.rest.search_utils \
            import GetValuesWithStorageManager

        workflow = self.get_workflow(deployment, workflow_id)
        workflow_parameters = workflow.get('parameters', {})
        custom_parameters = parameters.keys() - workflow_parameters.keys()
        if not self.allow_custom_parameters and custom_parameters:
            raise manager_exceptions.IllegalExecutionParametersError(
                f'Workflow "{workflow_id}" does not have the following '
                f'parameters declared: { ",".join(custom_parameters) }. '
                f'Remove these parameters or use the flag for allowing '
                f'custom parameters') from None

        wrong_types = {}
        for name, param in workflow_parameters.items():
            declared_type = param.get('type')
            if declared_type is None or name not in parameters:
                continue
            try:
                parameters[name] = self._convert_param_type(
                        parameters[name], declared_type)
            except ValueError:
                wrong_types[name] = declared_type
        if wrong_types:
            raise manager_exceptions.IllegalExecutionParametersError('\n'.join(
                f'Parameter "{n}" must be of type {t}'
                for n, t in wrong_types.items())
            )

        for name, param in workflow_parameters.items():
            if 'default' in param:
                parameters.setdefault(name, param['default'])

        constraint_violations = {}
        for name, param in workflow_parameters.items():
            constraints = extract_constraints(param)
            if name not in parameters:
                continue
            param_type = param.get('type')
            if param_type not in TYPES_BASED_ON_DB_ENTITIES \
                    and not constraints:
                continue
            try:
                getter = GetValuesWithStorageManager(get_storage_manager(),
                                                     self.deployment_id)
                validate_input_value(name, constraints, parameters[name],
                                     param_type, getter)
            except (dsl_exceptions.DSLParsingException,
                    dsl_exceptions.ConstraintException) as ex:
                constraint_violations[name] = ex
        if constraint_violations:
            raise manager_exceptions.IllegalExecutionParametersError('\n'.join(
                f'Parameter "{n}" does not meet its constraints: {e}'
                for n, e in constraint_violations.items()
            ))

        missing_parameters = workflow_parameters.keys() - parameters.keys()
        if missing_parameters:
            raise manager_exceptions.IllegalExecutionParametersError(
                f'Workflow "{workflow_id}" must be provided with the following'
                f' parameters to execute: { ",".join(missing_parameters) }'
            ) from None

    def _convert_param_type(self, param, target_type):
        if target_type == 'integer':
            return int(param)
        elif target_type == 'boolean':
{'fraction_non_alphanumeric': 0.08505129080952434, 'fraction_numerical': 0.000991659045583261, 'mean_word_length': 2.36178093862281, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 67, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '821826', 'n_tokens_mistral': 25530, 'n_tokens_neox': 24658, 'n_words': 4832}