Initial commit.

README.md

@@ -1,2 +1,3 @@
 # ml-pipeline-engineering
+
 Accompanies the (forthcoming) blog series on engineering ML pipeline.

flake8.ini

@@ -0,0 +1,6 @@
+[flake8]
+filename = *.py
+exclude = 
+    __pycache__
+max_line_length = 89
+ignore = W503,W605

mypy.ini

@@ -0,0 +1,13 @@
+[mypy]
+files = pipeline
+warn_return_any = True
+disallow_untyped_calls=True
+
+[mypy-uvicorn.*]
+ignore_missing_imports = True
+
+[mypy-pydantic.*]
+ignore_missing_imports = True
+
+[mypy-fastapi.*]
+ignore_missing_imports = True

notebooks/requirements_nb.txt

@@ -0,0 +1,7 @@
+jupyterlab==3.0.16
+seaborn==0.11.1
+numpy==1.21.0
+pandas==1.2.5
+scikit-learn==0.24.2
+boto3==1.17.101
+joblib==1.0.1

pipeline/serve_model.py

@@ -0,0 +1,34 @@
+"""
+- Get model and load into memory.
+- Start web API server.
+"""
+from typing import Dict, Union
+
+import uvicorn
+from fastapi import FastAPI, status
+from pydantic import BaseModel
+
+app = FastAPI(debug=False)
+
+
+class Data(BaseModel):
+    product_code: str
+    orders_placed: float
+
+
+class Prediction(BaseModel):
+    est_hours_to_dispatch: float
+    model_version: str
+
+
+@app.post(
+    "/api/v0.0.1/time_to_dispatch",
+    status_code=status.HTTP_200_OK,
+    response_model=Prediction,
+)
+def time_to_dispatch(data: Data) -> Dict[str, Union[str, float]]:
+    return {"est_hours_to_dispatch": 1.0, "model_version": "0.0.1"}
+
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", workers=1)

pipeline/train_model.py

@@ -0,0 +1,13 @@
+"""
+- Download dataset.
+- Train model.
+- Persist model.
+"""
+
+
+def main() -> None:
+    print("Hello from train_model stage.")
+
+
+if __name__ == "__main__":
+    main()

requirements_cicd.txt

@@ -0,0 +1,5 @@
+bodywork==2.1.6
+tox==3.23.1
+pytest==6.2.4
+mypy==0.910
+flake8==3.9.2

requirements_pipe.txt

@@ -0,0 +1,7 @@
+numpy==1.21.0
+pandas==1.2.5
+scikit-learn==0.24.2
+boto3==1.17.101
+joblib==1.0.1
+fastapi==0.65.2
+uvicorn==0.14.0

tests/test_serve_model.py

@@ -0,0 +1,18 @@
+"""
+Tests for web API.
+"""
+from fastapi.testclient import TestClient
+
+from pipeline.serve_model import app
+
+test_client = TestClient(app)
+
+
+def test_web_api_returns_valid_response_given_valid_data():
+    prediction_request = {"product_code": "SKU001", "orders_placed": 100}
+    prediction_response = test_client.post(
+        "/api/v0.0.1/time_to_dispatch", json=prediction_request
+    )
+    assert prediction_response.status_code == 200
+    assert "est_hours_to_dispatch" in prediction_response.json().keys()
+    assert "model_version" in prediction_response.json().keys()

tests/test_train_model.py

@@ -0,0 +1,12 @@
+"""
+Tests for model training stage.
+"""
+from _pytest.capture import CaptureFixture
+
+from pipeline.train_model import main
+
+
+def test_main_execution(capsys: CaptureFixture):
+    main()
+    stdout = capsys.readouterr().out
+    assert "Hello from train_model stage" in stdout

tox.ini

@@ -0,0 +1,12 @@
+[tox]
+envlist = py38-{unit_and_functional_tests,static_code_analysis}
+
+[testenv]
+skip_install = true
+deps = 
+    -rrequirements_cicd.txt
+    -rrequirements_pipe.txt
+commands = 
+    unit_and_functional_tests: pytest tests/ --disable-warnings {posargs}
+    static_code_analysis: mypy --config-file mypy.ini
+    static_code_analysis: flake8 --config flake8.ini pipeline