refactor(backend): Clean up Library & Store DB schema (#9774)

Distilled from #9541 to reduce the scope of that PR.

- Part of #9307

- ❗ Blocks #9786
  - ❗ Blocks #9541

### Changes 🏗️

- Fix `LibraryAgent` schema (for #9786)
- Fix relationships between `LibraryAgent`, `AgentGraph`, and
`AgentPreset`
  - Impose uniqueness constraint on `LibraryAgent`

- Rename things that are called `agent` that actually refer to a
`graph`/`agentGraph`
- Fix singular/plural forms in DB schema
- Simplify reference names of closely related objects (e.g.
`AgentGraph.AgentGraphExecutions` -> `AgentGraph.Executions`)

- Eliminate use of `# type: ignore` in DB statements
  - Add `typed` and `typed_cast` utilities to `backend.util.type`

### Checklist 📋

#### For code changes:
- [x] I have clearly listed my changes in the PR description
- [x] I have made a test plan
- [x] I have tested my changes according to the test plan:
- [x] CI static type checking (with all risky `# type: ignore` removed)
  - [x] Check that column references in views are updated

Author: Pwuts

autogpt_platform/backend/backend/data/credit.py

@@ -3,6 +3,7 @@
 from abc import ABC, abstractmethod
 from collections import defaultdict
 from datetime import datetime, timezone
+from typing import cast
 
 import stripe
 from autogpt_libs.utils.cache import thread_cached
@@ -18,6 +19,7 @@
     CreditRefundRequestCreateInput,
     CreditTransactionCreateInput,
     CreditTransactionWhereInput,
+    IntFilter,
 )
 from tenacity import retry, stop_after_attempt, wait_exponential
 
@@ -213,7 +215,7 @@ async def _get_credits(self, user_id: str) -> tuple[int, datetime]:
                 "userId": user_id,
                 "createdAt": {"lte": top_time},
                 "isActive": True,
-                "runningBalance": {"not": None},  # type: ignore
+                "runningBalance": cast(IntFilter, {"not": None}),
             },
             order={"createdAt": "desc"},
         )

autogpt_platform/backend/backend/data/execution.py

@@ -23,6 +23,7 @@
     AgentNodeExecutionInputOutput,
 )
 from prisma.types import (
+    AgentGraphExecutionCreateInput,
     AgentGraphExecutionWhereInput,
     AgentNodeExecutionCreateInput,
     AgentNodeExecutionInputOutputCreateInput,
@@ -121,15 +122,15 @@ class GraphExecution(GraphExecutionMeta):
 
     @staticmethod
     def from_db(_graph_exec: AgentGraphExecution):
-        if _graph_exec.AgentNodeExecutions is None:
+        if _graph_exec.NodeExecutions is None:
             raise ValueError("Node executions must be included in query")
 
         graph_exec = GraphExecutionMeta.from_db(_graph_exec)
 
         complete_node_executions = sorted(
             [
                 NodeExecutionResult.from_db(ne, _graph_exec.userId)
-                for ne in _graph_exec.AgentNodeExecutions
+                for ne in _graph_exec.NodeExecutions
                 if ne.executionStatus != ExecutionStatus.INCOMPLETE
             ],
             key=lambda ne: (ne.queue_time is None, ne.queue_time or ne.add_time),
@@ -181,15 +182,15 @@ class GraphExecutionWithNodes(GraphExecution):
 
     @staticmethod
     def from_db(_graph_exec: AgentGraphExecution):
-        if _graph_exec.AgentNodeExecutions is None:
+        if _graph_exec.NodeExecutions is None:
             raise ValueError("Node executions must be included in query")
 
         graph_exec_with_io = GraphExecution.from_db(_graph_exec)
 
         node_executions = sorted(
             [
                 NodeExecutionResult.from_db(ne, _graph_exec.userId)
-                for ne in _graph_exec.AgentNodeExecutions
+                for ne in _graph_exec.NodeExecutions
             ],
             key=lambda ne: (ne.queue_time is None, ne.queue_time or ne.add_time),
         )
@@ -220,21 +221,21 @@ class NodeExecutionResult(BaseModel):
     end_time: datetime | None
 
     @staticmethod
-    def from_db(execution: AgentNodeExecution, user_id: Optional[str] = None):
-        if execution.executionData:
+    def from_db(_node_exec: AgentNodeExecution, user_id: Optional[str] = None):
+        if _node_exec.executionData:
             # Execution that has been queued for execution will persist its data.
-            input_data = type_utils.convert(execution.executionData, dict[str, Any])
+            input_data = type_utils.convert(_node_exec.executionData, dict[str, Any])
         else:
             # For incomplete execution, executionData will not be yet available.
             input_data: BlockInput = defaultdict()
-            for data in execution.Input or []:
+            for data in _node_exec.Input or []:
                 input_data[data.name] = type_utils.convert(data.data, type[Any])
 
         output_data: CompletedBlockOutput = defaultdict(list)
-        for data in execution.Output or []:
+        for data in _node_exec.Output or []:
             output_data[data.name].append(type_utils.convert(data.data, type[Any]))
 
-        graph_execution: AgentGraphExecution | None = execution.AgentGraphExecution
+        graph_execution: AgentGraphExecution | None = _node_exec.GraphExecution
         if graph_execution:
             user_id = graph_execution.userId
         elif not user_id:
@@ -246,17 +247,17 @@ def from_db(execution: AgentNodeExecution, user_id: Optional[str] = None):
             user_id=user_id,
             graph_id=graph_execution.agentGraphId if graph_execution else "",
             graph_version=graph_execution.agentGraphVersion if graph_execution else 0,
-            graph_exec_id=execution.agentGraphExecutionId,
-            block_id=execution.AgentNode.agentBlockId if execution.AgentNode else "",
-            node_exec_id=execution.id,
-            node_id=execution.agentNodeId,
-            status=execution.executionStatus,
+            graph_exec_id=_node_exec.agentGraphExecutionId,
+            block_id=_node_exec.Node.agentBlockId if _node_exec.Node else "",
+            node_exec_id=_node_exec.id,
+            node_id=_node_exec.agentNodeId,
+            status=_node_exec.executionStatus,
             input_data=input_data,
             output_data=output_data,
-            add_time=execution.addedTime,
-            queue_time=execution.queuedTime,
-            start_time=execution.startedTime,
-            end_time=execution.endedTime,
+            add_time=_node_exec.addedTime,
+            queue_time=_node_exec.queuedTime,
+            start_time=_node_exec.startedTime,
+            end_time=_node_exec.endedTime,
         )
 
 
@@ -351,29 +352,29 @@ async def create_graph_execution(
         The id of the AgentGraphExecution and the list of ExecutionResult for each node.
     """
     result = await AgentGraphExecution.prisma().create(
-        data={
-            "agentGraphId": graph_id,
-            "agentGraphVersion": graph_version,
-            "executionStatus": ExecutionStatus.QUEUED,
-            "AgentNodeExecutions": {
-                "create": [  # type: ignore
-                    {
-                        "agentNodeId": node_id,
-                        "executionStatus": ExecutionStatus.QUEUED,
-                        "queuedTime": datetime.now(tz=timezone.utc),
-                        "Input": {
+        data=AgentGraphExecutionCreateInput(
+            agentGraphId=graph_id,
+            agentGraphVersion=graph_version,
+            executionStatus=ExecutionStatus.QUEUED,
+            NodeExecutions={
+                "create": [
+                    AgentNodeExecutionCreateInput(
+                        agentNodeId=node_id,
+                        executionStatus=ExecutionStatus.QUEUED,
+                        queuedTime=datetime.now(tz=timezone.utc),
+                        Input={
                             "create": [
                                 {"name": name, "data": Json(data)}
                                 for name, data in node_input.items()
                             ]
                         },
-                    }
+                    )
                     for node_id, node_input in nodes_input
                 ]
             },
-            "userId": user_id,
-            "agentPresetId": preset_id,
-        },
+            userId=user_id,
+            agentPresetId=preset_id,
+        ),
         include=GRAPH_EXECUTION_INCLUDE_WITH_NODES,
     )
 
@@ -600,7 +601,7 @@ async def get_node_execution_results(
         "agentGraphExecutionId": graph_exec_id,
     }
     if block_ids:
-        where_clause["AgentNode"] = {"is": {"agentBlockId": {"in": block_ids}}}
+        where_clause["Node"] = {"is": {"agentBlockId": {"in": block_ids}}}
     if statuses:
         where_clause["OR"] = [{"executionStatus": status} for status in statuses]
 

autogpt_platform/backend/backend/data/graph.py

@@ -1,7 +1,7 @@
 import logging
 import uuid
 from collections import defaultdict
-from typing import Any, Literal, Optional, Type
+from typing import Any, Literal, Optional, Type, cast
 
 import prisma
 from prisma import Json
@@ -10,7 +10,9 @@
 from prisma.types import (
     AgentGraphCreateInput,
     AgentGraphWhereInput,
+    AgentGraphWhereInputRecursive1,
     AgentNodeCreateInput,
+    AgentNodeIncludeFromAgentNodeRecursive1,
     AgentNodeLinkCreateInput,
 )
 from pydantic.fields import computed_field
@@ -465,13 +467,11 @@ def from_db(
             is_active=graph.isActive,
             name=graph.name or "",
             description=graph.description or "",
-            nodes=[
-                NodeModel.from_db(node, for_export) for node in graph.AgentNodes or []
-            ],
+            nodes=[NodeModel.from_db(node, for_export) for node in graph.Nodes or []],
             links=list(
                 {
                     Link.from_db(link)
-                    for node in graph.AgentNodes or []
+                    for node in graph.Nodes or []
                     for link in (node.Input or []) + (node.Output or [])
                 }
             ),
@@ -602,8 +602,8 @@ async def get_graph(
         and not (
             await StoreListingVersion.prisma().find_first(
                 where={
-                    "agentId": graph_id,
-                    "agentVersion": version or graph.version,
+                    "agentGraphId": graph_id,
+                    "agentGraphVersion": version or graph.version,
                     "isDeleted": False,
                     "submissionStatus": SubmissionStatus.APPROVED,
                 }
@@ -637,12 +637,16 @@ async def get_sub_graphs(graph: AgentGraph) -> list[AgentGraph]:
         sub_graph_ids = [
             (graph_id, graph_version)
             for graph in search_graphs
-            for node in graph.AgentNodes or []
+            for node in graph.Nodes or []
             if (
                 node.AgentBlock
                 and node.AgentBlock.id == agent_block_id
-                and (graph_id := dict(node.constantInput).get("graph_id"))
-                and (graph_version := dict(node.constantInput).get("graph_version"))
+                and (graph_id := cast(str, dict(node.constantInput).get("graph_id")))
+                and (
+                    graph_version := cast(
+                        int, dict(node.constantInput).get("graph_version")
+                    )
+                )
             )
         ]
         if not sub_graph_ids:
@@ -651,13 +655,16 @@ async def get_sub_graphs(graph: AgentGraph) -> list[AgentGraph]:
         graphs = await AgentGraph.prisma().find_many(
             where={
                 "OR": [
-                    {
-                        "id": graph_id,
-                        "version": graph_version,
-                        "userId": graph.userId,  # Ensure the sub-graph is owned by the same user
-                    }
+                    type_utils.typed(
+                        AgentGraphWhereInputRecursive1,
+                        {
+                            "id": graph_id,
+                            "version": graph_version,
+                            "userId": graph.userId,  # Ensure the sub-graph is owned by the same user
+                        },
+                    )
                     for graph_id, graph_version in sub_graph_ids
-                ]  # type: ignore
+                ]
             },
             include=AGENT_GRAPH_INCLUDE,
         )
@@ -671,7 +678,13 @@ async def get_sub_graphs(graph: AgentGraph) -> list[AgentGraph]:
 async def get_connected_output_nodes(node_id: str) -> list[tuple[Link, Node]]:
     links = await AgentNodeLink.prisma().find_many(
         where={"agentNodeSourceId": node_id},
-        include={"AgentNodeSink": {"include": AGENT_NODE_INCLUDE}},  # type: ignore
+        include={
+            "AgentNodeSink": {
+                "include": cast(
+                    AgentNodeIncludeFromAgentNodeRecursive1, AGENT_NODE_INCLUDE
+                )
+            }
+        },
     )
     return [
         (Link.from_db(link), NodeModel.from_db(link.AgentNodeSink))
@@ -829,12 +842,12 @@ async def fix_llm_provider_credentials():
             SELECT    graph."userId"       user_id,
                   node.id              node_id,
                   node."constantInput" node_preset_input
-        FROM      platform."AgentNode"  node
-        LEFT JOIN platform."AgentGraph" graph
-        ON        node."agentGraphId" = graph.id
-        WHERE     node."constantInput"::jsonb->'credentials'->>'provider' = 'llm'
-        ORDER BY  graph."userId";
-        """
+            FROM      platform."AgentNode"  node
+            LEFT JOIN platform."AgentGraph" graph
+            ON        node."agentGraphId" = graph.id
+            WHERE     node."constantInput"::jsonb->'credentials'->>'provider' = 'llm'
+            ORDER BY  graph."userId";
+            """
         )
         logger.info(f"Fixing LLM credential inputs on {len(broken_nodes)} nodes")
     except Exception as e: