Source code for forml.flow._suite.member

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you 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
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

Flow members represent partial pipeline blocks during pipeline assembly.
import abc
import typing
import weakref

from .. import _exception
from . import assembly

if typing.TYPE_CHECKING:
    from forml import flow

[docs]class Composable(metaclass=abc.ABCMeta): """Common interface for operators and expressions.""" def __rshift__(self, right: 'flow.Composable') -> 'Compound': """Semantical composition construct.""" return Compound(right, self) def __repr__(self): return self.__class__.__name__
[docs] @abc.abstractmethod def compose(self, scope: 'flow.Composable') -> 'flow.Trunk': """Implementation of the internal task graph and its composition with the preceding part of the expression. Args: scope: Preceding part of the expression that this operator is supposed to compose with. Returns: Trunk instance representing the composed task graph. """
[docs] @abc.abstractmethod def expand(self) -> 'flow.Trunk': """Compose this instance and the entire preceding part of the expression and return the resulting trunk. This is typically called by a downstream operator (*right* side of the expression) within its :meth:`compose` method where this is passed as part of the *left* side of the expression. Returns: Trunk instance representing the composed task graph. """
class Origin(Composable): """Initial builder without a predecessor.""" def compose(self, scope: 'flow.Composable') -> 'flow.Trunk': """Origin composition is just the left side trunk. Args: scope: Left side composable. Returns: Trunk. """ return scope.expand() def expand(self) -> 'flow.Trunk': return assembly.Trunk()
[docs]class Operator(Composable, metaclass=abc.ABCMeta): # pylint: disable=abstract-method """Base class for operator implementations.""" def expand(self) -> 'flow.Trunk': return self.compose(Origin())
class Compound(Composable): """Operator chaining descriptor.""" _TERMS = weakref.WeakValueDictionary() def __init__(self, right: 'flow.Composable', left: 'flow.Composable'): for term in (left, right): if not isinstance(term, Composable): raise ValueError(f'{type(term)} not composable') if term in self._TERMS: raise _exception.TopologyError(f'Non-linear {term} composition') self._TERMS[term] = self self._right: 'flow.Composable' = right self._left: 'flow.Composable' = left def __repr__(self): return f'{self._left} >> {self._right}' def compose(self, scope: 'flow.Composable') -> 'flow.Trunk': """Expression composition is just extension of its segments. Args: scope: Left side composable. Returns: Trunk. """ return scope.expand().extend(*self.expand()) def expand(self) -> 'flow.Trunk': return self._right.compose(self._left)