Module: successor_dag
Expand source code
# Copyright (C) 2023-present The Project Contributors
#
# 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
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from dataclasses import dataclass
from cl.runtime import Context
from cl.runtime import RecordMixin
from cl.runtime.records.dataclasses_extensions import missing
from cl.runtime.view.dag.dag import Dag
from cl.runtime.views.dag.successor_dag_key import SuccessorDagKey
from cl.runtime.views.dag.successor_dag_node import SuccessorDagNode
from cl.runtime.views.dag.successor_dag_node_key import SuccessorDagNodeKey
@dataclass(slots=True, kw_only=True)
class SuccessorDag(SuccessorDagKey, RecordMixin[SuccessorDagKey]):
"""Directed acyclic graph (DAG) where each node defines its successors."""
title: str = missing()
"""Title of the DAG."""
root_node: SuccessorDagNodeKey = missing()
"""Root node of the DAG."""
def get_key(self) -> SuccessorDagKey:
return SuccessorDagKey(dag_id=self.dag_id)
def view_dag(self) -> Dag | None:
"""DAG view."""
if self.root_node is None:
return None
root_node = Context.current().load_one(SuccessorDagNodeKey, self.root_node)
if root_node is None:
return None
return SuccessorDagNode.build_dag(node=root_node)
Classes
class SuccessorDag (*, dag_id: str = None, title: str = None, root_node: SuccessorDagNodeKey = None)
-
Directed acyclic graph (DAG) where each node defines its successors.
Expand source code
@dataclass(slots=True, kw_only=True) class SuccessorDag(SuccessorDagKey, RecordMixin[SuccessorDagKey]): """Directed acyclic graph (DAG) where each node defines its successors.""" title: str = missing() """Title of the DAG.""" root_node: SuccessorDagNodeKey = missing() """Root node of the DAG.""" def get_key(self) -> SuccessorDagKey: return SuccessorDagKey(dag_id=self.dag_id) def view_dag(self) -> Dag | None: """DAG view.""" if self.root_node is None: return None root_node = Context.current().load_one(SuccessorDagNodeKey, self.root_node) if root_node is None: return None return SuccessorDagNode.build_dag(node=root_node)
Ancestors
- SuccessorDagKey
- KeyMixin
- abc.ABC
- RecordMixin
- typing.Generic
Static methods
def get_key_type() -> Type
-
Inherited from:
SuccessorDagKey
.get_key_type
Return key type even when called from a record.
Fields
var dag_id -> str
-
Inherited from:
SuccessorDagKey
.dag_id
Unique DAG identifier.
var root_node -> SuccessorDagNodeKey
-
Root node of the DAG.
var title -> str
-
Title of the DAG.
Methods
def get_key(self) -> SuccessorDagKey
-
Inherited from:
RecordMixin
.get_key
Return a new key object whose fields populated from self, do not return self.
def init_all(self) -> None
-
Inherited from:
RecordMixin
.init_all
Invoke ‘init’ for each class in the order from base to derived, then validate against schema.
def view_dag(self) -> Dag | None
-
DAG view.