cosmic-cortex

version: "3"

services:
  fastapi-ml-quickstart:
    build: .
    ports:
      - 8000:8000
    entrypoint: ["pytest"]

cosmic-cortex

# This GitHub action describes the continuous integration workflow for pull requests to master.

name: ci

on:
  pull_request:
    branches:
      - master
  push:
    branches:

cosmic-cortex

version: "3"

services:
  fastapi-ml-quickstart:
    build: .
    ports:
      - 8000:8000

cosmic-cortex

FROM ubuntu:19.10

COPY ./api /api/api
COPY requirements.txt /requirements.txt

RUN apt-get update \
    && apt-get install python3-dev python3-pip -y \
    && pip3 install -r requirements.txt

ENV PYTHONPATH=/api

cosmic-cortex

import pandas as pd

from fastapi import File, UploadFile, HTTPException
from starlette.status import HTTP_422_UNPROCESSABLE_ENTITY


@app.post("/predict_csv")
def predict_csv(csv_file: UploadFile = File(...), model: Model = Depends(get_model)):
    try:
        df = pd.read_csv(csv_file.file).astype(float)

cosmic-cortex

from itertools import product

from starlette.status import HTTP_422_UNPROCESSABLE_ENTITY


@pytest.mark.parametrize(
    "n_instances, test_data_n_features",
    product(range(1, 10), [n for n in range(1, 20) if n != n_features]),
)
def test_predict_with_wrong_input(

cosmic-cortex

import pytest
import random

from starlette.testclient import TestClient
from starlette.status import HTTP_200_OK

from api.ml.model import n_features


@pytest.mark.parametrize("n_instances", range(1, 10))

cosmic-cortex

import pytest
from starlette.testclient import TestClient

from ..main import app

from ..ml.model import get_model
from .mocks import MockModel


def get_model_override():

cosmic-cortex

import numpy as np


class MockModel:
    def __init__(self, model_path: str = None):
        self._model_path = None
        self._model = None

    def predict(self, X: np.ndarray) -> np.ndarray:
        n_instances = len(X)

cosmic-cortex

from pydantic import BaseModel, ValidationError, validator

from .ml.model import n_features


class PredictRequest(BaseModel):
    data: List[List[float]]

    @validator("data")
    def check_dimensionality(cls, v):