Become the Maestro of your MLOps Abstractions

report = Report(metrics=[
    DataDriftPreset(), 
])

report.run(reference_data=reference, current_data=current)
report

import abc
import typing as T
import pydantic as pdt
from sklearn import ensemble, pipeline

ParamKey = str
ParamValue = T.Any
Params = T.Dict[ParamKey, ParamValue]

class Model(abc.ABC, pdt.BaseModel):
    """Base class for a model."""

    KIND: str

    def get_params(self, deep: bool = True) -> Params:
        """Get the model params."""
        params: Params = {}
        for key, value in self.dict().items():
            if not key.startswith("_") and not key.isupper():
                params[key] = value
        return params

    def set_params(self, **params: ParamValue) -> "Model":
        """Set the model params in place."""
        for key, value in params.items():
            setattr(self, key, value)
        return self

    @abc.abstractmethod
    def fit(self, inputs: schemas.Inputs, target: schemas.Target) -> "Model":
        """Fit the model on the given inputs and target."""

    @abc.abstractmethod
    def predict(self, inputs: schemas.Inputs) -> schemas.Output:
        """Generate an output with the model for the given inputs."""

import abc
import typing as T
import pydantic as pdt

Locals = T.Dict[str, T.Any]

class Job(abc.ABC, pdt.BaseModel):
    """Base class for a job."""

    KIND: str

    @abc.abstractmethod
    def run(self) -> Locals:
        """Run the job in context."""

class TrainingJob(Job):
    """Train and register a single AI/ML model"""

    KIND: T.Literal["TrainingJob"] = "TrainingJob"

    inputs: datasets.DatasetKind
    target: datasets.DatasetKind
    model: models.ModelKind = models.BaselineSklearnModel()
    metric: metrics.MetricKind = metrics.SklearnMetric()

    def run(self) -> Locals:
        """Run the training job in context."""
        # lots of code here ...
        return locals()

import typing as T
from cloudpathlib import AnyPath
from omegaconf import DictConfig, ListConfig, OmegaConf

Config = T.Union[ListConfig, DictConfig]

def load_config(path: str) -> Config:
    """Load a configuration file."""
    any_path = AnyPath(path)
    text = any_path.read_text()
    config = OmegaConf.create(text)
    return config