I have created a Flask app with Flask-appbuilder that has models and an API on top of one table.

I want to post multiple rows of data to the table through the API.

As per the FAB documentation I can add one row to the table by posting a JSON object, but I want to add more than one row in the same request. I keep getting the same error:

"_schema":["Invalid input type."]

The request payload is a list of dicts that looks something like this:

data = [
    {"sample_id": 1, "score_1": 0.1483, "score_2": 0.665047, "test_id": "102"},
    {"sample_id": 2, "score_1": 0.2255, "score_2": 0.688011, "test_id": "103"},
    {"sample_id": 3, "score_1": 0.161, "score_2": 0.672661, "test_id": "104"},
    {"sample_id": 4, "score_1": 0.1563, "score_2": 0.67177, "test_id": "105"},
    {"sample_id": 5, "score_1": 0.1958, "score_2": 0.677987, "test_id": "106"}
]

My models and API look like this:

from flask_appbuilder import Model
from flask_appbuilder.models.mixins import AuditMixin
from flask_appbuilder import ModelRestApi
from flask_appbuilder.models.sqla.interface import SQLAInterface
from sqlalchemy import (
    Column,
    Integer,
    String,
    DECIMAL,
    ForeignKey,
)
from sqlalchemy.orm import relationship


class Sample(AuditMixin, Model):
    id = Column(Integer, primary_key=True)
    sample_id = Column(String(512))


class TestResults(Model, AuditMixin):
    id = Column(Integer, primary_key=True)
    test_id = Column(String(512), nullable=False)
    sample_id = Column(Integer, ForeignKey("sample.id"))
    sample = relationship("Sample", foreign_keys=[sample_id])
    score_1 = Column(DECIMAL(4, 2), nullable=False)
    score_2 = Column(DECIMAL(4, 2), nullable=False)


class TestResultsApi(ModelRestApi):
    resource_name = "testresults"
    datamodel = SQLAInterface(TestResults)

    base_permissions = ["can_post", "can_get"]

The function I’m using to post to the API takes a dataframe and table name (“testresults”) as arguments. It looks like this:

import json
import logging
import requests


def post_data(base_url, token, headers, data, table_name):
    payload = []
    for row in data.itertuples(index=False):
        row_dict = row._asdict()
        payload.append(row_dict)
    print(payload)
    headers["Authorization"] = f"Bearer {token}"
    url = f"{self.base_url}/api/v1/{table_name}"
    resp = requests.post(
        url,
        data=json.dumps(payload),
        headers=headers,
    )
    print(resp)
    if resp.status_code >= 400:
        logging.warning("Request Failed")
        logging.warning(resp.text)

Any help greatly appreciated!