Basechainnode

laktory.models.BaseChainNode

Bases: BaseModel

Functions

get_sources

get_sources(cls=None)

Get all sources feeding the Chain Node

Source code in laktory/models/transformers/basechainnode.py

def get_sources(self, cls=None) -> list[BaseDataSource]:
    """Get all sources feeding the Chain Node"""

    from laktory.models.datasources.basedatasource import BaseDataSource

    if cls is None:
        cls = BaseDataSource

    sources = []
    for a in self.parsed_func_args:
        if isinstance(a.value, cls):
            sources += [a.value]
    for a in self.parsed_func_kwargs.values():
        if isinstance(a.value, cls):
            sources += [a.value]

    if self.sql_expr:
        sources += self.parsed_sql_expr.node_data_sources

    return sources