mltrace.entities package

Module contents

class mltrace.entities.Component(name: str = '', owner: str = '', description: str = '', beforeTests: list = [], afterTests: list = [], tags: List[str] = [])

Bases: mltrace.entities.base.Base

afterRun(**local_vars)

Computation to execute after running a component. Will run all test objects listed in afterTests.

property afterTests: list

beforeRun(**kwargs)

Computation to execute before running a component. Will run each test object listed in beforeTests.

property beforeTests: list

property description: str

property name: str

property owner: str

run(inputs: List[str] = [], outputs: List[str] = [], input_vars: List[str] = [], output_vars: List[str] = [], input_kwargs: Dict[str, str] = {}, output_kwargs: Dict[str, str] = {}, endpoint: bool = False, staleness_threshold: int = 2592000, auto_log: bool = False, *user_args, **user_kwargs)

Decorator around the function executed: c = Component() @c.run def my_function(arg1, arg2):

do_something()

arg1 and arg2 are the arguments passed to the beforeRun and afterRun methods. We first execute the beforeRun method, then the function itself, then the afterRun method with the values of the args at the end of the function.

ADD DESCRIPTION HERE ABOUT INPUT VARIABLEs and what they are

property tags: List[str]

class mltrace.entities.ComponentRun(component_name: str, notes: str = '', start_timestamp: Optional[datetime.datetime] = None, end_timestamp: Optional[datetime.datetime] = None, inputs: List[mltrace.entities.io_pointer.IOPointer] = [], outputs: List[mltrace.entities.io_pointer.IOPointer] = [], git_hash: Optional[str] = None, git_tags: Optional[List[str]] = None, code_snapshot: Optional[str] = None, id: Optional[str] = None, stale: List[str] = [], dependencies: List[str] = [])

Bases: mltrace.entities.base.Base

Component Run abstraction.

add_input(inp: Union[str, mltrace.entities.io_pointer.IOPointer], pointer_type: Optional[mltrace.db.models.PointerTypeEnum] = None)

Add a single input (instance of IOPointer).

add_inputs(inputs: List[Union[str, mltrace.entities.io_pointer.IOPointer]])

Add a list of inputs (each element should be an instance of IOPointer).

add_output(out: Union[str, mltrace.entities.io_pointer.IOPointer], pointer_type: Optional[mltrace.db.models.PointerTypeEnum] = None)

“Add a single output (instance of IOPointer).

add_outputs(outputs: List[Union[str, mltrace.entities.io_pointer.IOPointer]])

Add a list of outputs (each element should be an instance of IOPointer).

property code_snapshot: str

property component_name: str

property dependencies: List[str]

property end_timestamp: datetime.datetime

property git_hash: str

property git_tags: List[str]

property id: str

property inputs: List[mltrace.entities.io_pointer.IOPointer]

property notes: str

property outputs: List[mltrace.entities.io_pointer.IOPointer]

set_end_timestamp(ts: Optional[datetime.datetime] = None)

set_start_timestamp(ts: Optional[datetime.datetime] = None)

set_upstream(dependencies: Union[str, List[str]])

Set dependencies for this ComponentRun. API similar to Airflow set_upstream. It will grab the most recent run for the dependency name.

property stale: List[str]

property start_timestamp: datetime.datetime

class mltrace.entities.IOPointer(name: str, value: Any = '', pointer_type: mltrace.db.models.PointerTypeEnum = PointerTypeEnum.UNKNOWN, flag: bool = False)

Bases: mltrace.entities.base.Base

property flag: bool

property name: str

property pointer_type: mltrace.db.models.PointerTypeEnum

property value: Any

class mltrace.entities.Test(name: str = '')

Bases: object

getTestMethods()

Gets all methods in this class that start with “test”

property name

runTests(**kwargs)

Runs all tests in this class.