49 lines
1.1 KiB
Python
49 lines
1.1 KiB
Python
from collections.abc import Callable
|
|
from pydantic import BaseModel, computed_field
|
|
|
|
from plesna.models.storage import Table
|
|
|
|
|
|
class Transformation(BaseModel):
|
|
"""
|
|
The function have to have at least 2 arguments: sources and targets
|
|
Other arguments will came throught extra_kwrds
|
|
|
|
The function will have to return metadata as dict
|
|
"""
|
|
|
|
function: Callable
|
|
extra_kwrds: dict = {}
|
|
|
|
|
|
class Flux(BaseModel):
|
|
id: str
|
|
name: str
|
|
sources: list[Table]
|
|
targets: list[Table]
|
|
transformation: Transformation
|
|
|
|
@computed_field
|
|
@property
|
|
def sources_dict(self) -> dict[str, Table]:
|
|
return {s.id: s for s in self.sources}
|
|
|
|
@computed_field
|
|
@property
|
|
def sources_id(self) -> dict[str, Table]:
|
|
return [s.id for s in self.sources]
|
|
|
|
@computed_field
|
|
@property
|
|
def targets_id(self) -> dict[str, Table]:
|
|
return [s.id for s in self.targets]
|
|
|
|
@computed_field
|
|
@property
|
|
def targets_dict(self) -> dict[str, Table]:
|
|
return {s.id: s for s in self.targets}
|
|
|
|
|
|
class FluxMetaData(BaseModel):
|
|
data: dict
|