Source code for renku.core.workflow.activity

# Copyright 2017-2023 - Swiss Data Science Center (SDSC)
# A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
# Eidgenössische Technische Hochschule Zürich (ETHZ).
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.
"""Activity management."""

import itertools
import os
from collections import defaultdict
from pathlib import Path
from typing import Dict, FrozenSet, Iterable, List, NamedTuple, Optional, Set, Tuple

import networkx
from pydantic import validate_arguments

from renku.command.command_builder import inject
from renku.core import errors
from renku.core.interface.activity_gateway import IActivityGateway
from renku.core.util import communication
from renku.core.workflow.plan import get_activities, is_plan_removed, remove_plan
from renku.domain_model.entity import Entity
from renku.domain_model.project_context import project_context
from renku.domain_model.provenance.activity import Activity, Usage

[docs]def get_activities_until_paths( paths: List[str], sources: List[str], activity_gateway: IActivityGateway, revision: Optional[str] = None ) -> Set[Activity]: """Get all current activities leading to `paths`, from `sources`.""" all_activities: Dict[str, Set[Activity]] = defaultdict(set) def include_newest_activity(activity): existing_activities = all_activities[] add_activity_if_recent(activity=activity, activities=existing_activities) repository = project_context.repository commit = None if revision: commit = repository.get_commit(revision) for path in paths: checksum = None if commit: try: blob = commit.tree[path] except KeyError: raise errors.GitError(f"Couldn't find file {path} at revision {revision}") checksum = blob.hexsha activities = activity_gateway.get_activities_by_generation(path, checksum=checksum) if len(activities) == 0: communication.warn(f"Path '{path}' is not generated by any workflows.") continue latest_activity = max(activities, key=lambda a: a.ended_at_time) upstream_chains = activity_gateway.get_upstream_activity_chains(latest_activity) if sources: # NOTE: Add the activity to check if it also matches the condition upstream_chains.append((latest_activity,)) # NOTE: Only include paths that is using at least one of the sources upstream_chains = [c for c in upstream_chains if any(u.entity.path in sources for u in c[-1].usages)] # NOTE: Include activity only if any of its upstream match the condition if upstream_chains: include_newest_activity(latest_activity) else: include_newest_activity(latest_activity) for chain in upstream_chains: for activity in chain: include_newest_activity(activity) return {a for activities in all_activities.values() for a in activities}
[docs]def create_activity_graph( activities: List[Activity], remove_overridden_parents=True, with_inputs_outputs=False, with_hidden_dependencies: bool = False, ) -> networkx.Graph: """Create a dependency DAG from activities.""" by_usage: Dict[str, Set[Activity]] = defaultdict(set) by_generation: Dict[str, Set[Activity]] = defaultdict(set) overridden_activities: Dict[Activity, Set[str]] = defaultdict(set) graph = networkx.DiGraph() def connect_nodes_based_on_dependencies(): for activity in activities: # NOTE: Make sure that activity is in the graph in case it has no connection to others graph.add_node(activity) if with_inputs_outputs: create_input_output_edges(activity) collection = ( itertools.chain(activity.usages, activity.hidden_usages) if with_hidden_dependencies else activity.usages ) for usage in collection: path = usage.entity.path by_usage[path].add(activity) parent_activities = by_generation[path] for parent in parent_activities: create_edge(parent, activity, path) for generation in activity.generations: path = generation.entity.path by_generation[path].add(activity) child_activities = by_usage[path] for child in child_activities: create_edge(activity, child, path) def create_input_output_edges(activity): for generation in activity.generations: path = generation.entity.path if not graph.has_node(path): graph.add_node(path) if not graph.has_edge(activity, path): graph.add_edge(activity, path) collection = ( itertools.chain(activity.usages, activity.hidden_usages) if with_hidden_dependencies else activity.usages ) for usage in collection: path = usage.entity.path if not graph.has_node(path): graph.add_node(path) if not graph.has_edge(path, activity): graph.add_edge(path, activity) def create_edge(parent, child, path: str): if with_inputs_outputs: if not graph.has_edge(parent, path): graph.add_edge(parent, path) if not graph.has_edge(path, child): graph.add_edge(path, child) else: if graph.has_edge(parent, child): return graph.add_edge(parent, child, path=path) def connect_nodes_by_execution_order(): for path, values in by_generation.items(): if len(values) <= 1: continue # NOTE: Order multiple activities that generate a common path create_order_among_activities(values, path) def create_order_among_activities(activities: Set[Activity], path): for a, b in itertools.combinations(activities, 2): if (networkx.has_path(graph, a, b) and path in overridden_activities[a]) or ( networkx.has_path(graph, b, a) and path in overridden_activities[b] ): continue # NOTE: More recent activity should be executed after the other one # NOTE: This won't introduce a cycle in the graph because there is no other path between the two nodes comparison = a.compare_to(b) if comparison < 0: if not networkx.has_path(graph, a, b): graph.add_edge(a, b) overridden_activities[a].add(path) elif comparison > 0: if not networkx.has_path(graph, b, a): graph.add_edge(b, a) overridden_activities[b].add(path) else: raise ValueError(f"Cannot create an order between activities that generate '{path}': {a} and {b}") def remove_overridden_activities(): to_be_removed = set() to_be_processed = set(overridden_activities.keys()) while len(to_be_processed) > 0: activity = to_be_processed.pop() overridden_paths = overridden_activities[activity] generated_path = {g.entity.path for g in activity.generations} if generated_path != overridden_paths: continue # NOTE: All generated paths are overridden; there is no point in executing the activity to_be_removed.add(activity) if not remove_overridden_parents: continue # NOTE: Check if its parents can be removed as well for parent in graph.predecessors(activity): if parent in to_be_removed: continue data = graph.get_edge_data(parent, activity) if data and "path" in data: overridden_activities[parent].add(data["path"]) to_be_processed.add(parent) for activity in to_be_removed: graph.remove_node(activity) connect_nodes_based_on_dependencies() cycles = list(networkx.algorithms.cycles.simple_cycles(graph)) if cycles: cycles = [map(lambda x: getattr(x, "id", x), cycle) for cycle in cycles] raise errors.GraphCycleError(cycles) connect_nodes_by_execution_order() remove_overridden_activities() return graph
[docs]def sort_activities(activities: List[Activity], remove_overridden_parents=True) -> List[Activity]: """Return a sorted list of activities based on their dependencies and execution order.""" graph = create_activity_graph(activities, remove_overridden_parents) return list(networkx.topological_sort(graph))
[docs]class ModifiedActivitiesEntities(NamedTuple): """A class containing sets of modified/deleted activities and entities for both normal and hidden entities.""" modified: Set[Tuple[Activity, Entity]] """Set of modified activity and entity tuples.""" deleted: Set[Tuple[Activity, Entity]] """Set of deleted activity and entity tuples.""" hidden_modified: Set[Tuple[Activity, Entity]] """Set of modified activity and entity tuples for hidden entities."""
[docs]@inject.autoparams("activity_gateway") def get_all_modified_and_deleted_activities_and_entities( repository, activity_gateway: IActivityGateway, check_hidden_dependencies: bool = False ) -> ModifiedActivitiesEntities: """ Return latest activities with at least one modified or deleted input along with the modified/deleted input entity. An activity can be repeated if more than one of its inputs are modified. Args: repository: The current ``Repository``. activity_gateway(IActivityGateway): The injected Activity gateway. Returns: ModifiedActivitiesEntities: Modified and deleted activities and entities. """ all_activities = activity_gateway.get_all_activities() relevant_activities = filter_overridden_activities(all_activities) return get_modified_activities( activities=relevant_activities, repository=repository, check_hidden_dependencies=check_hidden_dependencies )
[docs]@inject.autoparams() def get_downstream_generating_activities( starting_activities: Set[Activity], paths: List[str], ignore_deleted: bool, project_path: Path, activity_gateway: IActivityGateway, ) -> List[Activity]: """Return activities downstream of passed activities that generate at least a path in ``paths``. Args: starting_activities(Set[Activity]): Activities to use as starting/upstream nodes. paths(List[str]): Optional generated paths to end downstream chains at. ignore_deleted(bool): Whether to ignore deleted generations. project_path(Path): Path to project's root directory. activity_gateway(IActivityGateway): The injected Activity gateway. Returns: Set[Activity]: All activities and their downstream activities. """ all_activities: Dict[str, Set[Activity]] = defaultdict(set) def include_newest_activity(activity): existing_activities = all_activities[] add_activity_if_recent(activity=activity, activities=existing_activities) def does_activity_generate_any_paths(activity) -> bool: is_same = any(g.entity.path in paths for g in activity.generations) is_parent = any(Path(p) in Path(g.entity.path).parents for p in paths for g in activity.generations) return is_same or is_parent def has_an_existing_generation(activity) -> bool: for generation in activity.generations: if (project_path / generation.entity.path).exists(): return True return False for starting_activity in starting_activities: downstream_chains = activity_gateway.get_downstream_activity_chains(starting_activity) if paths: # NOTE: Add the activity to check if it also matches the condition downstream_chains.append((starting_activity,)) downstream_chains = [c for c in downstream_chains if does_activity_generate_any_paths(c[-1])] # NOTE: Include activity only if any of its downstream matched the condition include_starting_activity = len(downstream_chains) > 0 elif ignore_deleted: # NOTE: Excluded deleted generations only if they are not passed in ``paths`` # NOTE: Add the activity to check if it also matches the condition downstream_chains.append((starting_activity,)) downstream_chains = [c for c in downstream_chains if has_an_existing_generation(c[-1])] # NOTE: Include activity only if any of its downstream matched the condition include_starting_activity = len(downstream_chains) > 0 else: include_starting_activity = True if include_starting_activity: include_newest_activity(starting_activity) for chain in downstream_chains: for activity in chain: if not is_activity_valid(activity): # don't process further downstream activities as the plan in question was deleted break include_newest_activity(activity) return list({a for activities in all_activities.values() for a in activities})
[docs]def get_modified_activities( activities: FrozenSet[Activity], repository, check_hidden_dependencies: bool ) -> ModifiedActivitiesEntities: """Get lists of activities that have modified/deleted usage entities.""" def get_modified_activities_helper(hashes, modified, deleted, hidden: bool): for activity in activities: collection: List[Usage] = activity.hidden_usages if hidden else activity.usages # type: ignore for usage in collection: entity = usage.entity current_checksum = hashes.get(entity.path, None) usage_path = repository.path / usage.entity.path if current_checksum is None or not usage_path.exists(): deleted.add((activity, entity)) elif current_checksum != entity.checksum: modified.add((activity, entity)) modified: Set[Tuple[Activity, Entity]] = set() deleted: Set[Tuple[Activity, Entity]] = set() hidden_modified: Set[Tuple[Activity, Entity]] = set() paths = [] hidden_paths = [] for activity in activities: for usage in activity.usages: paths.append(usage.entity.path) if check_hidden_dependencies: for usage in activity.hidden_usages: hidden_paths.append(usage.entity.path) hashes = repository.get_object_hashes(paths=paths) get_modified_activities_helper(hashes=hashes, modified=modified, deleted=deleted, hidden=False) if check_hidden_dependencies and hidden_paths: hashes = repository.get_object_hashes(paths=hidden_paths) get_modified_activities_helper(hashes=hashes, modified=hidden_modified, deleted=set(), hidden=True) return ModifiedActivitiesEntities(modified=modified, deleted=deleted, hidden_modified=hidden_modified)
[docs]def filter_overridden_activities(activities: List[Activity]) -> FrozenSet[Activity]: """Filter out overridden activities from a list of activities.""" relevant_activities: Dict[FrozenSet[str], Activity] = {} for activity in activities[::-1]: outputs = frozenset(g.entity.path for g in activity.generations) subset_of = set() superset_of = set() for o, a in relevant_activities.items(): if outputs.issubset(o): subset_of.add((o, a)) elif outputs.issuperset(o): superset_of.add((o, a)) if not subset_of and not superset_of: relevant_activities[outputs] = activity continue if subset_of and any(activity.ended_at_time < a.ended_at_time for _, a in subset_of): # activity is a subset of another, newer activity, ignore it continue older_subsets = [o for o, a in superset_of if activity.ended_at_time > a.ended_at_time] for older_subset in older_subsets: # remove other activities that this activity is a superset of del relevant_activities[older_subset] relevant_activities[outputs] = activity return frozenset(relevant_activities.values())
[docs]def add_activity_if_recent(activity: Activity, activities: Set[Activity]): """Add ``activity`` to ``activities`` if it's not in the set or is the latest executed instance. Remove existing activities that were executed earlier. """ if activity in activities: return for existing_activity in activities: if activity.has_identical_inputs_and_outputs_as(existing_activity): if activity.ended_at_time > existing_activity.ended_at_time: # activity is newer activities.remove(existing_activity) activities.add(activity) return # NOTE: No similar activity was found activities.add(activity)
[docs]def get_latest_activity(activities: Iterable[Activity]) -> Optional[Activity]: """Return the activity that was executed after all other activities.""" return max(activities, key=lambda a: a.ended_at_time) if activities else None
[docs]def get_latest_activity_before(activities: Iterable[Activity], activity: Activity) -> Optional[Activity]: """Return the latest activity that was executed before the passed activity.""" activities_before = [a for a in activities if a.ended_at_time <= activity.ended_at_time and !=] return get_latest_activity(activities_before)
[docs]@inject.autoparams("activity_gateway") @validate_arguments(config=dict(arbitrary_types_allowed=True)) def revert_activity( *, activity_gateway: IActivityGateway, activity_id: str, delete_plan: bool, force: bool, metadata_only: bool ) -> Activity: """Revert an activity. Args: activity_gateway(IActivityGateway): The injected activity gateway. activity_id(str): ID of the activity to be reverted. delete_plan(bool): Delete the plan if it's not used by any other activity. force(bool): Revert the activity even if it has some downstream activities. metadata_only(bool): Only revert the metadata and don't touch generated files. Returns: The deleted activity. """ repository = project_context.repository def delete_associated_plan(activity): if not delete_plan: return plan = activity.association.plan used_by_other_activities = any(a for a in get_activities(plan) if != if used_by_other_activities: return remove_plan(, force=True) def revert_generations(activity) -> Tuple[Set[str], Set[str]]: """Either revert each generation to an older version (created by an earlier activity) or delete it.""" deleted_paths = set() updated_paths: Dict[str, str] = {} if metadata_only: return set(), set() for generation in activity.generations: path = generation.entity.path generator_activities = activity_gateway.get_activities_by_generation(path=path) generator_activities = [a for a in generator_activities if is_activity_valid(a) and not a.deleted] latest_generator = get_latest_activity(generator_activities) if latest_generator != activity: # NOTE: A newer activity already generated the same path continue previous_generator = get_latest_activity_before(generator_activities, activity) if previous_generator is None: # NOTE: The activity is the only generator # NOTE: Delete the path if there are no downstreams otherwise keep it downstream_activities = activity_gateway.get_activities_by_usage(path) if not downstream_activities: deleted_paths.add(path) elif not force: raise errors.ActivityDownstreamNotEmptyError(activity) else: # NOTE: There is a previous generation of that path, so, revert to it previous_generation = next(g for g in previous_generator.generations if g.entity.path == path) updated_paths[path] = previous_generation.entity.checksum for path, previous_checksum in updated_paths.items(): try: repository.copy_content_to_file(path, checksum=previous_checksum, output_path=path) except errors.FileNotFound: communication.warn(f"Cannot revert '{path}' to a previous version, will keep the current version") for path in deleted_paths: try: os.unlink(project_context.path / path) except OSError: communication.warn(f"Cannot delete '{path}'") return deleted_paths, set(updated_paths.keys()) activity = activity_gateway.get_by_id(activity_id) if activity is None: raise errors.ParameterError(f"Cannot find activity with ID '{activity}'") if activity.deleted: raise errors.ParameterError(f"Activity with ID '{activity}' is already deleted") # NOTE: The order of removal is important here so don't change it delete_associated_plan(activity) revert_generations(activity) activity_gateway.remove(activity, force=force) # NOTE: Delete the activity after processing metadata or otherwise we won't see the activity as the latest generator activity.delete() return activity
[docs]def is_activity_valid(activity: Activity) -> bool: """Return whether this plan has not been deleted. Args: activity(Activity): The Activity whose Plan should be checked. Returns: bool: True if the activities' Plan is still valid, False otherwise. """ return not is_plan_removed(plan=activity.association.plan)