Set a nice name and an MLOps deployment, get a nice app

mlops_deployment_app.py

import json
import requests

from h2o_wave import main, app, Q, ui, expando_to_dict


ENDPOINT = "https://model.DOMAIN.h2o.ai/MODELID/model"
APP_TITLE = "My Use Case"


@app("/")
async def serve(q: Q):

    if not q.client.initialized:

        q.client.model_endpoint = ENDPOINT

        sample_response = mlops_get_schema(q.client.model_endpoint)
        q.client.model_fields = sample_response["fields"]
        q.client.single_row = sample_response["rows"][0]

        sample_prediction_response = mlops_get_single_prediction(
            q.client.model_endpoint, q.client.model_fields, q.client.single_row
        )
        q.client.prediction_fields = sample_prediction_response["fields"]
        q.client.single_prediction_value = sample_prediction_response["score"][0]

        single_prediction_details = [
            ui.text(
                f"**{q.client.prediction_fields[i]}**: {q.client.single_prediction_value[i]}"
            )
            for i in range(len(q.client.prediction_fields))
        ]

        single_prediction_ui_items = [
            ui.inline(
                items=[
                    ui.text_xl("Single Predictions"),
                    ui.button(name="score", label="Make Prediction", primary=True),
                ],
                inset=False,
                justify="between",
            ),
            ui.inline(items=single_prediction_details, inset=True),
        ]

        for i in range(len(q.client.model_fields)):
            single_prediction_ui_items.append(
                ui.textbox(
                    name=f"col_{q.client.model_fields[i]}",
                    label=q.client.model_fields[i],
                    value=q.client.single_row[i],
                )
            )

        q.page["meta"] = ui.meta_card(
            box="",
            title=APP_TITLE,
            theme="ember",
            layouts=[
                ui.layout(
                    breakpoint="xl",
                    width="1200px",
                    zones=[
                        ui.zone(
                            "main",
                            size="100vh",
                            zones=[
                                ui.zone("header"),
                                ui.zone("body", size="1"),
                                ui.zone("footer"),
                            ],
                        )
                    ],
                )
            ],
        )
        q.page["header"] = ui.header_card(
            box="header", title=APP_TITLE, subtitle="Predictions on models"
        )
        q.page["single_prediction"] = ui.form_card(
            box="body", items=single_prediction_ui_items
        )
        q.page["footer"] = ui.footer_card(box="footer", caption="Made with H2O Wave.")

        q.client.initialized = True

    if q.args.score:
        user_input_dictionary = expando_to_dict(q.args)
        print(user_input_dictionary)

        for key in user_input_dictionary:
            if key.startswith("col_"):
                col_name = key.lstrip("col_")
                col_index = q.client.model_fields.index(col_name)
                q.client.single_row[col_index] = user_input_dictionary[key]

        prediction_response = mlops_get_single_prediction(
            q.client.model_endpoint, q.client.model_fields, q.client.single_row
        )
        q.client.prediction_fields = prediction_response["fields"]
        q.client.single_prediction_value = prediction_response["score"][0]

        q.page["single_prediction"].items[1].inline.items = [
            ui.text(
                f"**{q.client.prediction_fields[i]}**: {q.client.single_prediction_value[i]}"
            )
            for i in range(len(q.client.prediction_fields))
        ]

    await q.page.save()


def mlops_get_schema(endpoint_url):
    url = endpoint_url + "/sample_request"
    response = requests.get(url)

    # Return response if status is valid
    if response.status_code == 200:
        return json.loads(response.text)


def mlops_get_single_prediction(endpoint_url, model_fields, data_row):
    url = endpoint_url + "/score"
    post_dictionary = {"fields": model_fields, "rows": [data_row]}
    response = requests.post(url=url, json=post_dictionary)

    # Return response if status is valid
    if response.status_code == 200:
        return json.loads(response.text)