Skip to content

sklab.logging

sklab.logging.NoOpLogger dataclass

Logger that drops all logging calls.

Useful as the default logger when no external tracking backend is configured.

Source code in src/sklab/_logging/noop.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@dataclass
class NoOpLogger:
    """Logger that drops all logging calls.

    Useful as the default logger when no external tracking backend is configured.
    """

    @contextmanager
    def start_run(self, name=None, config=None, tags=None, nested=False):
        yield self

    def log_params(self, params) -> None:
        pass

    def log_metrics(self, metrics, step: int | None = None) -> None:
        pass

    def set_tags(self, tags) -> None:
        pass

    def log_artifact(self, path: str, name: str | None = None) -> None:
        pass

    def log_model(self, model: Any, name: str | None = None) -> None:
        pass

sklab.logging.MLflowLogger dataclass

Logger that tracks experiments with MLflow.

MLflow uses module-level functions that operate on the active run, so we don't need to store run state.

Source code in src/sklab/_logging/mlflow.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@dataclass
class MLflowLogger:
    """Logger that tracks experiments with MLflow.

    MLflow uses module-level functions that operate on the active run,
    so we don't need to store run state.
    """

    experiment_name: str | None = None

    @contextmanager
    def start_run(self, name=None, config=None, tags=None, nested=False):
        if self.experiment_name:
            mlflow.set_experiment(self.experiment_name)
        with mlflow.start_run(run_name=name, nested=nested):
            if config:
                self.log_params(config)
            if tags:
                self.set_tags(tags)
            yield self

    def log_params(self, params) -> None:
        mlflow.log_params(dict(params))

    def log_metrics(self, metrics, step: int | None = None) -> None:
        mlflow.log_metrics(dict(metrics), step=step)

    def set_tags(self, tags) -> None:
        mlflow.set_tags(dict(tags))

    def log_artifact(self, path: str, name: str | None = None) -> None:
        if name is None:
            mlflow.log_artifact(path)
        else:
            mlflow.log_artifact(path, name=name)

    def log_model(self, model: Any, name: str | None = None) -> None:
        mlflow.sklearn.log_model(model, name=name or "model")

sklab.logging.WandbLogger dataclass

Logger that tracks experiments with Weights & Biases.

W&B requires calling methods on the run object, so we store a reference to the active run in self._run.

Source code in src/sklab/_logging/wandb.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@dataclass
class WandbLogger:
    """Logger that tracks experiments with Weights & Biases.

    W&B requires calling methods on the run object, so we store
    a reference to the active run in `self._run`.
    """

    project: str | None = None
    entity: str | None = None
    _run: Any = field(default=None, init=False, repr=False)

    @contextmanager
    def start_run(self, name=None, config=None, tags=None, nested=False):
        with wandb.init(
            project=self.project,
            entity=self.entity,
            name=name,
            config=config or {},
            tags=list(tags.values()) if tags else None,
            reinit=nested,
        ) as run:
            self._run = run
            yield self
        self._run = None

    def log_params(self, params) -> None:
        self._run.config.update(dict(params), allow_val_change=True)

    def log_metrics(self, metrics, step: int | None = None) -> None:
        self._run.log(dict(metrics), step=step)

    def set_tags(self, tags) -> None:
        existing = set(self._run.tags or [])
        self._run.tags = sorted(existing | set(tags.values()))

    def log_artifact(self, path: str, name: str | None = None) -> None:
        artifact = wandb.Artifact(name or "artifact", type="file")
        artifact.add_file(path)
        self._run.log_artifact(artifact)

    def log_model(self, model: Any, name: str | None = None) -> None:
        if isinstance(model, str):
            self.log_artifact(model, name=name or "model")