Repository: PrefectHQ/prefect
Pull request: Add data model and methods for FlowRunInput
This adds the `FlowRunInput` data model, schemas and CRUD methods to support the human-in-the-loop work.

Closes #11337 
### Checklist
<!-- These boxes may be checked after opening the pull request. -->

- [x] This pull request references any related issue by including "closes `<link to issue>`"
	- If no issue exists and your change is not a small fix, please [create an issue](https://github.com/PrefectHQ/prefect/issues/new/choose) first.
- [x] This pull request includes tests or only affects documentation.
- [x] This pull request includes a label categorizing the change e.g. `fix`, `feature`, `enhancement`, `docs`.
  <!-- If you do not have permission to add a label, a maintainer will add one for you -->

For documentation changes:

- [ ] This pull request includes redirect settings in `netlify.toml` for files that are removed or renamed


Problem statement:
Add modeling for FlowRunInput to OSS
For human-in-the-loop we need a way to store key/value data, to that end we should add modeling for `FlowRunInput` that enables us to store key/value information associated with a flow run. We'll then use this data store to build out the interactivity piece.

Hints from issue discussion:

Commit message: Add data model and methods for FlowRunInput

Patch to review:
diff --git a/src/prefect/server/database/interface.py b/src/prefect/server/database/interface.py
index 4e4ae402e144..0f2e49aa3a6a 100644
--- a/src/prefect/server/database/interface.py
+++ b/src/prefect/server/database/interface.py
@@ -258,6 +258,11 @@ def Variable(self):
         """A variable model"""
         return self.orm.Variable
 
+    @property
+    def FlowRunInput(self):
+        """A flow run input model"""
+        return self.orm.FlowRunInput
+
     @property
     def deployment_unique_upsert_columns(self):
         """Unique columns for upserting a Deployment"""
diff --git a/src/prefect/server/database/migrations/MIGRATION-NOTES.md b/src/prefect/server/database/migrations/MIGRATION-NOTES.md
index eeb371fe491a..832b9f992df0 100644
--- a/src/prefect/server/database/migrations/MIGRATION-NOTES.md
+++ b/src/prefect/server/database/migrations/MIGRATION-NOTES.md
@@ -8,6 +8,10 @@ Each time a database migration is written, an entry is included here with:
 
 This gives us a history of changes and will create merge conflicts if two migrations are made at once, flagging situations where a branch needs to be updated before merging.
 
+# Add `flow_run_input` table.
+SQLite: `f5c3aaa499e8`
+Postgres: `8f83fc6d55eb`
+
 # Add last_polled to deployment table
 SQLite: `f3165ae0a213`
 Postgres: `bfe653bbf62e`
diff --git a/src/prefect/server/database/migrations/versions/postgresql/2023_12_06_102745_8f83fc6d55eb_create_flow_run_input_table.py b/src/prefect/server/database/migrations/versions/postgresql/2023_12_06_102745_8f83fc6d55eb_create_flow_run_input_table.py
new file mode 100644
index 000000000000..8b55b3737cbc
--- /dev/null
+++ b/src/prefect/server/database/migrations/versions/postgresql/2023_12_06_102745_8f83fc6d55eb_create_flow_run_input_table.py
@@ -0,0 +1,63 @@
+"""Create `flow_run_input` table
+
+Revision ID: 8f83fc6d55eb
+Revises: 9c493c02ca6d
+Create Date: 2023-12-06 10:27:45.646091
+
+"""
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import Text
+
+import prefect
+
+# revision identifiers, used by Alembic.
+revision = "8f83fc6d55eb"
+down_revision = "9c493c02ca6d"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "flow_run_input",
+        sa.Column(
+            "id",
+            prefect.server.utilities.database.UUID(),
+            server_default=sa.text("(GEN_RANDOM_UUID())"),
+            nullable=False,
+        ),
+        sa.Column(
+            "created",
+            prefect.server.utilities.database.Timestamp(timezone=True),
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated",
+            prefect.server.utilities.database.Timestamp(timezone=True),
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+            nullable=False,
+        ),
+        sa.Column("key", sa.String(), nullable=False),
+        sa.Column(
+            "value",
+            prefect.server.utilities.database.JSON(astext_type=Text()),
+            nullable=False,
+        ),
+        sa.Column(
+            "flow_run_id", prefect.server.utilities.database.UUID(), nullable=False
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_flow_run_input")),
+        sa.UniqueConstraint(
+            "flow_run_id", "key", name=op.f("uq_flow_run_input__flow_run_id_key")
+        ),
+    )
+    op.create_index(
+        op.f("ix_flow_run_input__updated"), "flow_run_input", ["updated"], unique=False
+    )
+
+
+def downgrade():
+    op.drop_index(op.f("ix_flow_run_input__updated"), table_name="flow_run_input")
+    op.drop_table("flow_run_input")
diff --git a/src/prefect/server/database/migrations/versions/sqlite/2023_12_06_102931_f5c3aaa499e8_create_flow_run_input_table.py b/src/prefect/server/database/migrations/versions/sqlite/2023_12_06_102931_f5c3aaa499e8_create_flow_run_input_table.py
new file mode 100644
index 000000000000..97d2d668c9fe
--- /dev/null
+++ b/src/prefect/server/database/migrations/versions/sqlite/2023_12_06_102931_f5c3aaa499e8_create_flow_run_input_table.py
@@ -0,0 +1,73 @@
+"""Create `flow_run_input` table
+
+Revision ID: f5c3aaa499e8
+Revises: 9c493c02ca6d
+Create Date: 2023-12-06 10:29:31.106059
+
+"""
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import Text
+
+import prefect
+
+# revision identifiers, used by Alembic.
+revision = "f5c3aaa499e8"
+down_revision = "9c493c02ca6d"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "flow_run_input",
+        sa.Column(
+            "id",
+            prefect.server.utilities.database.UUID(),
+            server_default=sa.text(
+                "(\n    (\n        lower(hex(randomblob(4)))\n        || '-'\n       "
+                " || lower(hex(randomblob(2)))\n        || '-4'\n        ||"
+                " substr(lower(hex(randomblob(2))),2)\n        || '-'\n        ||"
+                " substr('89ab',abs(random()) % 4 + 1, 1)\n        ||"
+                " substr(lower(hex(randomblob(2))),2)\n        || '-'\n        ||"
+                " lower(hex(randomblob(6)))\n    )\n    )"
+            ),
+            nullable=False,
+        ),
+        sa.Column(
+            "created",
+            prefect.server.utilities.database.Timestamp(timezone=True),
+            server_default=sa.text("(strftime('%Y-%m-%d %H:%M:%f000', 'now'))"),
+            nullable=False,
+        ),
+        sa.Column(
+            "updated",
+            prefect.server.utilities.database.Timestamp(timezone=True),
+            server_default=sa.text("(strftime('%Y-%m-%d %H:%M:%f000', 'now'))"),
+            nullable=False,
+        ),
+        sa.Column("key", sa.String(), nullable=False),
+        sa.Column(
+            "value",
+            prefect.server.utilities.database.JSON(astext_type=Text()),
+            nullable=False,
+        ),
+        sa.Column(
+            "flow_run_id", prefect.server.utilities.database.UUID(), nullable=False
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_flow_run_input")),
+        sa.UniqueConstraint(
+            "flow_run_id", "key", name=op.f("uq_flow_run_input__flow_run_id_key")
+        ),
+    )
+    with op.batch_alter_table("flow_run_input", schema=None) as batch_op:
+        batch_op.create_index(
+            batch_op.f("ix_flow_run_input__updated"), ["updated"], unique=False
+        )
+
+
+def downgrade():
+    with op.batch_alter_table("flow_run_input", schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f("ix_flow_run_input__updated"))
+
+    op.drop_table("flow_run_input")
diff --git a/src/prefect/server/database/orm_models.py b/src/prefect/server/database/orm_models.py
index 4a04f338350f..c5139e8dd5d1 100644
--- a/src/prefect/server/database/orm_models.py
+++ b/src/prefect/server/database/orm_models.py
@@ -1339,6 +1339,21 @@ class ORMVariable:
     __table_args__ = (sa.UniqueConstraint("name"),)
 
 
+@declarative_mixin
+class ORMFlowRunInput:
+    @declared_attr
+    def flow_run_id(cls):
+        return sa.Column(
+            UUID(),
+            nullable=False,
+        )
+
+    key = sa.Column(sa.String, nullable=False)
+    value = sa.Column(JSON, nullable=False)
+
+    __table_args__ = (sa.UniqueConstraint("flow_run_id", "key"),)
+
+
 class BaseORMConfiguration(ABC):
     """
     Abstract base class used to inject database-specific ORM configuration into Prefect.
@@ -1398,6 +1413,7 @@ def __init__(
         agent_mixin=ORMAgent,
         configuration_mixin=ORMConfiguration,
         variable_mixin=ORMVariable,
+        flow_run_input_mixin=ORMFlowRunInput,
     ):
         self.base_metadata = base_metadata or sa.schema.MetaData(
             # define naming conventions for our Base class to use
@@ -1451,6 +1467,7 @@ def __init__(
             block_document_reference_mixin=block_document_reference_mixin,
             configuration_mixin=configuration_mixin,
             variable_mixin=variable_mixin,
+            flow_run_input_mixin=flow_run_input_mixin,
         )
 
     def _unique_key(self) -> Tuple[Hashable, ...]:
@@ -1500,6 +1517,7 @@ def _create_orm_models(
         agent_mixin=ORMAgent,
         configuration_mixin=ORMConfiguration,
         variable_mixin=ORMVariable,
+        flow_run_input_mixin=ORMFlowRunInput,
     ):
         """
         Defines the ORM models used in Prefect REST API and binds them to the `self`. This method
@@ -1584,6 +1602,9 @@ class Configuration(configuration_mixin, self.Base):
         class Variable(variable_mixin, self.Base):
             pass
 
+        class FlowRunInput(flow_run_input_mixin, self.Base):
+            pass
+
         self.Flow = Flow
         self.FlowRunState = FlowRunState
         self.TaskRunState = TaskRunState
@@ -1610,6 +1631,7 @@ class Variable(variable_mixin, self.Base):
         self.FlowRunNotificationQueue = FlowRunNotificationQueue
         self.Configuration = Configuration
         self.Variable = Variable
+        self.FlowRunInput = FlowRunInput
 
     @property
     @abstractmethod
diff --git a/src/prefect/server/models/__init__.py b/src/prefect/server/models/__init__.py
index 8b96a4cd4917..35e52f1cd8b4 100644
--- a/src/prefect/server/models/__init__.py
+++ b/src/prefect/server/models/__init__.py
@@ -9,6 +9,7 @@
     concurrency_limits_v2,
     configuration,
     deployments,
+    flow_run_input,
     flow_run_notification_policies,
     flow_run_states,
     flow_runs,
diff --git a/src/prefect/server/models/flow_run_input.py b/src/prefect/server/models/flow_run_input.py
new file mode 100644
index 000000000000..12f23f0325b6
--- /dev/null
+++ b/src/prefect/server/models/flow_run_input.py
@@ -0,0 +1,118 @@
+import copy
+import uuid
+from typing import Optional
+
+import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.ext.asyncio import AsyncSession
+
+import prefect.server.schemas as schemas
+from prefect.server.database.dependencies import inject_db
+from prefect.server.database.interface import PrefectDBInterface
+
+# As of 2023-12-06 we're only planning on storing a single value per
+# FlowRunInput, however we're storing the value as a list of items to
+# allow for future expansion.
+VALUE_CONTAINER = {"__prefect_metadata": {}, "items": []}
+
+
+@inject_db
+async def create_flow_run_input(
+    session: AsyncSession,
+    db: PrefectDBInterface,
+    flow_run_input: schemas.actions.FlowRunInputCreate,
+) -> schemas.core.FlowRunInput:
+    data = flow_run_input.dict()
+
+    container = copy.deepcopy(VALUE_CONTAINER)
+    container["items"].append(
+        {
+            "_item_id": str(uuid.uuid4()),
+            "value": data["value"],
+        }
+    )
+    data["value"] = container
+
+    model = db.FlowRunInput(**data)
+    session.add(model)
+    await session.flush()
+
+    return schemas.core.FlowRunInput.from_orm(model)
+
+
+@inject_db
+async def read_flow_run_input(
+    session: AsyncSession,
+    db: PrefectDBInterface,
+    flow_run_id: uuid.UUID,
+    key: str,
+) -> Optional[schemas.core.FlowRunInput]:
+    query = sa.select(db.FlowRunInput).where(
+        sa.and_(
+            db.FlowRunInput.flow_run_id == flow_run_id,
+            db.FlowRunInput.key == key,
+        )
+    )
+
+    result = await session.execute(query)
+    model = result.scalar()
+    if model:
+        return schemas.core.FlowRunInput.from_orm(model)
+
+    return None
+
+
+@inject_db
+async def update_flow_run_input(
+    session: AsyncSession,
+    db: PrefectDBInterface,
+    flow_run_input: schemas.actions.FlowRunInputUpdate,
+) -> bool:
+    def set_value(new_value: str):
+        if db.dialect.name == "sqlite":
+            return sa.func.json_set(
+                db.FlowRunInput.value, "$.items[0].value", new_value
+            )
+        else:
+            return sa.func.jsonb_set(
+                db.FlowRunInput.value,
+                sa.text("'{items,0,value}'"),
+                sa.cast(new_value, JSONB),
+                True,
+            )
+
+    flow_run_id = flow_run_input.flow_run_id
+    key = flow_run_input.key
+    value = flow_run_input.value
+
+    query = (
+        sa.update(db.FlowRunInput)
+        .where(
+            sa.and_(
+                db.FlowRunInput.flow_run_id == flow_run_id,
+                db.FlowRunInput.key == key,
+            )
+        )
+        .values(value=set_value(value))
+    )
+
+    result = await session.execute(query)
+    return result.rowcount > 0
+
+
+@inject_db
+async def delete_flow_run_input(
+    session: AsyncSession,
+    db: PrefectDBInterface,
+    flow_run_id: uuid.UUID,
+    key: str,
+) -> bool:
+    result = await session.execute(
+        sa.delete(db.FlowRunInput).where(
+            sa.and_(
+                db.FlowRunInput.flow_run_id == flow_run_id, db.FlowRunInput.key == key
+            )
+        )
+    )
+
+    return result.rowcount > 0
diff --git a/src/prefect/server/schemas/actions.py b/src/prefect/server/schemas/actions.py
index e0e2b7b8750b..39256a816742 100644
--- a/src/prefect/server/schemas/actions.py
+++ b/src/prefect/server/schemas/actions.py
@@ -746,3 +746,25 @@ class VariableUpdate(ActionBaseModel):
 
     # validators
     _validate_name_format = validator("name", allow_reuse=True)(validate_variable_name)
+
+
+@copy_model_fields
+class FlowRunInputCreate(ActionBaseModel):
+    """Data used by the Prefect REST API to create a FlowRunInput"""
+
+    flow_run_id: UUID = FieldFrom(schemas.core.FlowRunInput)
+    key: str = FieldFrom(schemas.core.FlowRunInput)
+    value: str = Field(
+        description="The JSON value that should be stored for the input."
+    )
+
+
+@copy_model_fields
+class FlowRunInputUpdate(ActionBaseModel):
+    """Data used by the Prefect REST API to update a FlowRunInput"""
+
+    flow_run_id: UUID = FieldFrom(schemas.core.FlowRunInput)
+    key: str = FieldFrom(schemas.core.FlowRunInput)
+    value: str = Field(
+        description="The JSON value that should be stored for the input."
+    )
diff --git a/src/prefect/server/schemas/core.py b/src/prefect/server/schemas/core.py
index 4d9bcd792e09..7a8acb7380fe 100644
--- a/src/prefect/server/schemas/core.py
+++ b/src/prefect/server/schemas/core.py
@@ -1276,3 +1276,18 @@ class Variable(ORMBaseModel):
         description="A list of variable tags",
         example=["tag-1", "tag-2"],
     )
+
+
+class FlowRunInput(ORMBaseModel):
+    flow_run_id: UUID = Field(description="The flow run ID associated with the input.")
+    key: str = Field(description="The key of the input.")
+    value: Dict[str, Any] = Field(description="The container for the value.")
+
+    @property
+    def first_value(self) -> str:
+        return self.value["items"][0]
+
+    @validator("key", check_fields=False)
+    def validate_name_characters(cls, v):
+        raise_on_name_with_banned_characters(v)
+        return v
diff --git a/tests/server/models/test_flow_run_input.py b/tests/server/models/test_flow_run_input.py
new file mode 100644
index 000000000000..c4cde3f912f2
--- /dev/null
+++ b/tests/server/models/test_flow_run_input.py
@@ -0,0 +1,172 @@
+import json
+
+import pytest
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+import prefect.server.schemas as schemas
+from prefect.server import models
+
+
+class TestCreateFlowRunInput:
+    async def test_creates_flow_run_input(self, session: AsyncSession, flow_run):
+        flow_run_input = await models.flow_run_input.create_flow_run_input(
+            session=session,
+            flow_run_input=schemas.actions.FlowRunInputCreate(
+                flow_run_id=flow_run.id, key="my-key", value=json.dumps("my-value")
+            ),
+        )
+
+        assert isinstance(flow_run_input, schemas.core.FlowRunInput)
+        assert flow_run_input.flow_run_id == flow_run.id
+        assert flow_run_input.key == "my-key"
+
+        # The value is stored as a JSON string, and the modeling makes no
+        # attempts to decode it, so the value returned within the value
+        # container is a string that has been JSON encoded.
+        value_id = flow_run_input.value["items"][0]["_item_id"]
+        assert flow_run_input.value == {
+            "__prefect_metadata": {},
+            "items": [{"_item_id": value_id, "value": '"my-value"'}],
+        }
+
+    async def test_creates_flow_run_input_complex_value(
+        self, session: AsyncSession, flow_run
+    ):
+        flow_run_json = schemas.core.FlowRun.from_orm(flow_run).json()
+
+        flow_run_input = await models.flow_run_input.create_flow_run_input(
+            session=session,
+            flow_run_input=schemas.actions.FlowRunInputCreate(
+                flow_run_id=flow_run.id,
+                key="my-key",
+                value=flow_run_json,
+            ),
+        )
+
+        assert isinstance(flow_run_input, schemas.core.FlowRunInput)
+        assert flow_run_input.flow_run_id == flow_run.id
+        assert flow_run_input.key == "my-key"
+
+        value_id = flow_run_input.value["items"][0]["_item_id"]
+        assert flow_run_input.value == {
+            "__prefect_metadata": {},
+            "items": [{"_item_id": value_id, "value": flow_run_json}],
+        }
+
+    async def test_flow_run_id_and_key_are_unique(
+        self, session: AsyncSession, flow_run
+    ):
+        await models.flow_run_input.create_flow_run_input(
+            session=session,
+            flow_run_input=schemas.actions.FlowRunInputCreate(
+                flow_run_id=flow_run.id, key="my-key", value="my-value"
+            ),
+        )
+
+        with pytest.raises(IntegrityError):
+            await models.flow_run_input.create_flow_run_input(
+                session=session,
+                flow_run_input=schemas.actions.FlowRunInputCreate(
+                    flow_run_id=flow_run.id, key="my-key", value="my-value"
+                ),
+            )
+
+
+class TestReadFlowRunInput:
+    async def test_reads_flow_run_input(self, session: AsyncSession, flow_run):
+        await models.flow_run_input.create_flow_run_input(
+            session=session,
+            flow_run_input=schemas.actions.FlowRunInputCreate(
+                flow_run_id=flow_run.id, key="my-key", value="my-value"
+            ),
+        )
+
+        flow_run_input = await models.flow_run_input.read_flow_run_input(
+            session=session, flow_run_id=flow_run.id, key="my-key"
+        )
+
+        assert isinstance(flow_run_input, schemas.core.FlowRunInput)
+        assert flow_run_input.flow_run_id == flow_run.id
+        assert flow_run_input.key == "my-key"
+
+
+class TestUpdateFlowRun:
+    async def test_updates_flow_run_input(self, session: AsyncSession, flow_run):
+        await models.flow_run_input.create_flow_run_input(
+            session=session,
+            flow_run_input=schemas.actions.FlowRunInputCreate(
+                flow_run_id=flow_run.id, key="my-key", value=json.dumps("my-value")
+            ),
+        )
+
+        assert await models.flow_run_input.update_flow_run_input(
+            session=session,
+            flow_run_input=schemas.actions.FlowRunInputUpdate(
+                flow_run_id=flow_run.id,
+                key="my-key",
+                value=json.dumps("my-new-value"),
+            ),
+        )
+
+        flow_run_input = await models.flow_run_input.read_flow_run_input(
+            session=session, flow_run_id=flow_run.id, key="my-key"
+        )
+        value_id = flow_run_input.value["items"][0]["_item_id"]
+        assert flow_run_input.value == {
+            "__prefect_metadata": {},
+            "items": [{"_item_id": value_id, "value": '"my-new-value"'}],
+        }
+
+    async def test_updates_flow_run_input_complex_value(
+        self, session: AsyncSession, flow_run
+    ):
+        flow_run_input = await models.flow_run_input.create_flow_run_input(
+            session=session,
+            flow_run_input=schemas.actions.FlowRunInputCreate(
+                flow_run_id=flow_run.id,
+                key="my-key",
+                value="some-value",
+            ),
+        )
+
+        flow_run_json = schemas.core.FlowRun.from_orm(flow_run).json()
+
+        assert await models.flow_run_input.update_flow_run_input(
+            session=session,
+            flow_run_input=schemas.actions.FlowRunInputUpdate(
+                flow_run_id=flow_run.id,
+                key="my-key",
+                value=flow_run_json,
+            ),
+        )
+
+        flow_run_input = await models.flow_run_input.read_flow_run_input(
+            session=session, flow_run_id=flow_run.id, key="my-key"
+        )
+        value_id = flow_run_input.value["items"][0]["_item_id"]
+        assert flow_run_input.value == {
+            "__prefect_metadata": {},
+            "items": [{"_item_id": value_id, "value": flow_run_json}],
+        }
+
+
+class TestDeleteFlowRunInput:
+    async def test_deletes_flow_run_input(self, session: AsyncSession, flow_run):
+        await models.flow_run_input.create_flow_run_input(
+            session=session,
+            flow_run_input=schemas.actions.FlowRunInputCreate(
+                flow_run_id=flow_run.id, key="my-key", value="my-value"
+            ),
+        )
+
+        assert await models.flow_run_input.delete_flow_run_input(
+            session=session, flow_run_id=flow_run.id, key="my-key"
+        )
+
+        assert (
+            await models.flow_run_input.read_flow_run_input(
+                session=session, flow_run_id=flow_run.id, key="my-key"
+            )
+            is None
+        )


Code review comments:
diff_hunk: @@ -0,0 +1,118 @@
+import copy
+import uuid
+from typing import Optional
+
+import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.ext.asyncio import AsyncSession
+
+import prefect.server.schemas as schemas
+from prefect.server.database.dependencies import inject_db
+from prefect.server.database.interface import PrefectDBInterface
+
+# As of 2023-12-06 we're only planning on storing a single value per
+# FlowRunInput, however we're storing the value as a list of items to
+# allow for future expansion.
+VALUE_CONTAINER = {"__prefect_metadata": {}, "items": []}
+
+
+@inject_db
+async def create_flow_run_input(
+    session: AsyncSession,
+    db: PrefectDBInterface,
+    flow_run_input: schemas.actions.FlowRunInputCreate,
+) -> schemas.core.FlowRunInput:
+    data = flow_run_input.dict()
+
+    container = copy.deepcopy(VALUE_CONTAINER)
+    container["items"].append(
+        {
+            "_item_id": str(uuid.uuid4()),
+            "value": data["value"],
+        }
+    )
+    data["value"] = container
+
+    model = db.FlowRunInput(**data)
+    session.add(model)
+    await session.flush()
+
+    return schemas.core.FlowRunInput.from_orm(model)
+
+
+@inject_db
+async def read_flow_run_input(
+    session: AsyncSession,
+    db: PrefectDBInterface,
+    flow_run_id: uuid.UUID,
+    key: str,
+) -> Optional[schemas.core.FlowRunInput]:
+    query = sa.select(db.FlowRunInput).where(
+        sa.and_(
+            db.FlowRunInput.flow_run_id == flow_run_id,
+            db.FlowRunInput.key == key,
+        )
+    )
+
+    result = await session.execute(query)
+    model = result.scalar()
+    if model:
+        return schemas.core.FlowRunInput.from_orm(model)
+
+    return None
+
+
+@inject_db
+async def update_flow_run_input(
+    session: AsyncSession,
+    db: PrefectDBInterface,
+    flow_run_input: schemas.actions.FlowRunInputUpdate,
+) -> bool:
path: src/prefect/server/models/flow_run_input.py
text: @user1:
Is there a use case for updating an input?  I assumed that providing the input immediately set the state of the flow run to allow it to resume, so modifying the input seems like it could lead to inconsistency or fudging what really happened?

@author:
For pause/resume, not so much, I'd be okay with removing it for now even.

diff_hunk: @@ -746,3 +746,25 @@ class VariableUpdate(ActionBaseModel):
 
     # validators
     _validate_name_format = validator("name", allow_reuse=True)(validate_variable_name)
+
+
+@copy_model_fields
+class FlowRunInputCreate(ActionBaseModel):
+    """Data used by the Prefect REST API to create a FlowRunInput"""
+
+    flow_run_id: UUID = FieldFrom(schemas.core.FlowRunInput)
+    key: str = FieldFrom(schemas.core.FlowRunInput)
+    value: str = Field(
+        description="The JSON value that should be stored for the input."
+    )
path: src/prefect/server/schemas/actions.py
text: @user1:
Hmm, why only a `str` here?  I thought it would support JSON types (str, int, float, bool, dict, list)?

@author:
The server is agnostic about the serialization of the value that the user sending, it's just always treated as a string.