Reference library

ML engineering pipelines

Feature prep, batch inference, model-serving hooks, and production ML workflow glue.

5 matches

How to Mock Cron Schedule in Python

Compute the next scheduled run time for a cron expression using a pure-Python mock parser.

import re
from datetime import datetime, timedelta

class CronMock:
    def __init__(self, expression):
        self.expression = expression
        self.minutes = self._parse_field(expression.split()[0], 0, 59)
        self.hours = self._parse_field(expression.split()[1], 0, 23)
        self.days = self._parse_field(…

17 0 Open

ML engineering pipelines medium

How to Mock Kedro Pipeline Nodes in Python

Create a modular Kedro pipeline with node functions, namespacing, and input/output mapping to mock pipeline execution locally.

kedro pipeline modular

Python

from kedro.pipeline import Pipeline, node
from kedro.pipeline.modular_pipeline import pipeline as modular_pipeline


def preprocess(data: list) -> list:
    """Clean data by removing None values."""
    return [item for item in data if item is not None]


def transform(data: list) -> list:
    """Add 1 to each numeric…

15 0 Open

ML engineering pipelines medium

How to Mock ROC AUC in Python

Compute ROC AUC from scratch in Python using pairwise comparisons between positive and negative score distributions, ideal for testing ML models without sklearn.

machine-learning model-evaluation auc

Python

import random
from math import comb


def mock_roc_auc(scores, labels):
    """Compute mock ROC AUC by simulating a classifier's score distribution."""
    random.seed(42)
    n = len(labels)
    pos_scores = [scores[i] for i in range(n) if labels[i] == 1]
    neg_scores = [scores[i] for i in range(n) if labels[i] == …

12 0 Open

ML engineering pipelines medium

K-Fold Cross Validation in Python: A Simple Implementation

Implements k-fold cross validation from scratch, splitting data into folds and computing MSE scores for a baseline mean-predictor model.

cross-validation ml model-evaluation

Python

import random
from statistics import mean


def cross_validation_scores(data, labels, k=5, seed=42):
    random.seed(seed)
    indices = list(range(len(data)))
    random.shuffle(indices)
    fold_size = len(indices) // k
    folds = []
    for i in range(k):
        if i == k - 1:
            folds.append(indices[i *…

16 0 Open

ML engineering pipelines medium

Mock a Flyte ML workflow in Python

Build a lightweight mock of a Flyte ML pipeline with dataclasses and a simple execution loop that passes outputs between tasks.

flyte ml-pipeline dataclass

Python

from dataclasses import dataclass, field
from typing import List, Dict, Optional
import time


@dataclass
class FlyteTask:
    name: str
    inputs: Dict = field(default_factory=dict)
    outputs: Dict = field(default_factory=dict)

    def run(self) -> Dict:
        time.sleep(0.1)  # simulate work
        return sel…

16 0 Open

Browse by section

Each section groups closely related Python snippets.