Allow changes to Spec

codex/api.py

@@ -1,4 +1,3 @@
-import json
 import logging
 
 from fastapi import APIRouter, Path, Query
@@ -82,7 +81,7 @@ async def get_app(
         return app_response
     else:
         return JSONResponse(
-            content=json.dumps({"error": "Application not found"}),
+            content={"error": "Application not found"},
             status_code=404,
         )
 
@@ -105,7 +104,7 @@ async def delete_app(user_id: str, app_id: str):
     """
     await codex.database.delete_app(user_id, app_id)
     return JSONResponse(
-        content=json.dumps({"message": "Application deleted successfully"}),
+        content={"message": "Application deleted successfully"},
         status_code=200,
     )
 

codex/api_model.py

@@ -3,10 +3,12 @@
 from typing import List, Optional
 
 import prisma
-from prisma.enums import Role
-from prisma.models import Specification
+from prisma.enums import AccessLevel, Role
+from prisma.models import ObjectField, ObjectType, Specification
 from pydantic import BaseModel, Field
 
+from codex.common.parse_prisma import parse_prisma_schema
+
 logger = logging.getLogger(__name__)
 
 
@@ -147,24 +149,92 @@ class SpecificationCreate(BaseModel):
     description: str
 
 
+class SpecificationUpdate(prisma.models.Specification, BaseModel):
+    apiRouteSpecs: List[APIRouteSpecModel] = []
+
+
+class InputParamModel(BaseModel):
+    name: str
+    description: str
+    param_type: str
+
+
+class InputRequestResponseModel(BaseModel):
+    name: str
+    description: str
+    params: List[InputParamModel] = []
+
+
+class SpecificationAddRouteToModule(BaseModel):
+    function_name: str
+    access_level: AccessLevel
+    allowed_access_roles: List[str]
+    method: str
+    path: str
+    description: str
+    requestObject: Optional["ObjectTypeModel"] = None
+    responseObject: Optional["ObjectTypeModel"] = None
+
+
+class DatabaseEnums(BaseModel):
+    name: str
+    description: str
+    values: list[str]
+    definition: str
+
+    def __str__(self):
+        return f"**Enum: {self.name}**\n\n**Values**:\n{', '.join(self.values)}\n"
+
+
+class DatabaseTable(BaseModel):
+    name: str | None = None
+    description: str
+    definition: str  # prisma model for a table
+
+    def __str__(self):
+        return f"**Table: {self.name}**\n\n\n\n**Definition**:\n```\n{self.definition}\n```\n"
+
+
+class DatabaseSchema(BaseModel):
+    name: str  # name of the database schema
+    description: str  # context on what the database schema is
+    tables: List[DatabaseTable]  # list of tables in the database schema
+    enums: List[DatabaseEnums]
+
+    def __str__(self):
+        tables_str = "\n".join(str(table) for table in self.tables)
+        enum_str = "\n".join(str(enum) for enum in self.enums)
+        return f"## {self.name}\n**Description**: {self.description}\n**Tables**:\n{tables_str}\n**Enums**:\n{enum_str}\n"
+
+
+class ModuleWrapper(BaseModel):
+    id: str
+    name: str
+    description: str
+    interactions: str
+    apiRouteSpecs: List[APIRouteSpecModel] = []
+
+
 class SpecificationResponse(BaseModel):
     id: str
     createdAt: datetime
     name: str
     context: str
-    apiRouteSpecs: List[APIRouteSpecModel] = []
+    modules: List[ModuleWrapper] = []
+    databaseSchema: Optional[DatabaseSchema] = None
 
     @staticmethod
     def from_specification(specification: Specification) -> "SpecificationResponse":
         logger.debug(specification.model_dump_json())
-        routes = []
+        module_out = []
         modules: list[prisma.models.Module] | None = (
             specification.Modules if specification.Modules else None
         )
         if modules is None:
             raise ValueError("No routes found for the specification")
         for module in modules:
             if module.ApiRouteSpecs:
+                routes = []
                 for route in module.ApiRouteSpecs:
                     routes.append(
                         APIRouteSpecModel(
@@ -211,13 +281,65 @@ def from_specification(specification: Specification) -> "SpecificationResponse":
                             else None,
                         )
                     )
-
+                module_out.append(
+                    ModuleWrapper(
+                        id=module.id,
+                        apiRouteSpecs=routes,
+                        name=module.name,
+                        description=module.description,
+                        interactions=module.interactions,
+                    )
+                )
+            else:
+                module_out.append(
+                    ModuleWrapper(
+                        id=module.id,
+                        name=module.name,
+                        description=module.description,
+                        interactions=module.interactions,
+                    )
+                )
+        db_schema = None
+        if specification.DatabaseSchema:
+
+            def convert_to_table(table: prisma.models.DatabaseTable) -> DatabaseTable:
+                return DatabaseTable(
+                    name=table.name or "ERROR: Unknown Table Name",
+                    description=table.description,
+                    definition=table.definition,
+                )
+
+            def convert_to_enum(table: prisma.models.DatabaseTable) -> DatabaseEnums:
+                return DatabaseEnums(
+                    name=table.name or "ERROR: Unknown ENUM Name",
+                    description=table.description,
+                    values=parse_prisma_schema(table.definition)
+                    .enums[table.name or "ERROR: Unknown ENUM Name"]
+                    .values,
+                    definition=table.definition,
+                )
+
+            db_schema = DatabaseSchema(
+                name=specification.DatabaseSchema.name or "Database Schema",
+                tables=[
+                    convert_to_table(table)
+                    for table in specification.DatabaseSchema.DatabaseTables or []
+                    if not table.isEnum
+                ],
+                enums=[
+                    convert_to_enum(table)
+                    for table in specification.DatabaseSchema.DatabaseTables or []
+                    if table.isEnum
+                ],
+                description=specification.DatabaseSchema.description,
+            )
         ret_obj = SpecificationResponse(
             id=specification.id,
             createdAt=specification.createdAt,
             name="",
             context="",
-            apiRouteSpecs=routes,
+            modules=module_out,
+            databaseSchema=db_schema,
         )
 
         return ret_obj
@@ -272,3 +394,64 @@ class DeploymentResponse(DeploymentMetadata):
 class DeploymentsListResponse(BaseModel):
     deployments: List[DeploymentMetadata]
     pagination: Optional[Pagination] = None
+
+
+class ObjectTypeModel(BaseModel):
+    name: str = Field(description="The name of the object")
+    code: Optional[str] = Field(description="The code of the object", default=None)
+    description: Optional[str] = Field(
+        description="The description of the object", default=None
+    )
+    Fields: List["ObjectFieldModel"] = Field(description="The fields of the object")
+    is_pydantic: bool = Field(
+        description="Whether the object is a pydantic model", default=True
+    )
+    is_implemented: bool = Field(
+        description="Whether the object is implemented", default=True
+    )
+    is_enum: bool = Field(description="Whether the object is an enum", default=False)
+
+    def __init__(self, db_obj: ObjectType | None = None, **data):
+        if not db_obj:
+            super().__init__(**data)
+            return
+
+        super().__init__(
+            name=db_obj.name,
+            code=db_obj.code,
+            description=db_obj.description,
+            is_pydantic=db_obj.isPydantic,
+            is_enum=db_obj.isEnum,
+            Fields=[ObjectFieldModel(db_obj=f) for f in db_obj.Fields or []],
+            **data,
+        )
+
+
+class ObjectFieldModel(BaseModel):
+    name: str = Field(description="The name of the field")
+    description: Optional[str] = Field(
+        description="The description of the field", default=None
+    )
+    type: str = Field(
+        description="The type of the field. Can be a string like List[str] or an use any of they related types like list[User]",
+    )
+    value: Optional[str] = Field(description="The value of the field", default=None)
+    related_types: Optional[List[ObjectTypeModel]] = Field(
+        description="The related types of the field", default=[]
+    )
+
+    def __init__(self, db_obj: ObjectField | None = None, **data):
+        if not db_obj:
+            super().__init__(**data)
+            return
+
+        super().__init__(
+            name=db_obj.name,
+            description=db_obj.description,
+            type=db_obj.typeName,
+            value=db_obj.value,
+            related_types=[
+                ObjectTypeModel(db_obj=t) for t in db_obj.RelatedTypes or []
+            ],
+            **data,
+        )

codex/common/logging.py

@@ -1,4 +1,3 @@
-import json
 import logging
 import traceback
 from typing import Callable
@@ -52,7 +51,7 @@ async def execute_and_log(
         )
 
         return JSONResponse(
-            content=json.dumps({"error": str(e)}),
+            content={"error": str(e)},
             status_code=status_code,
         )
 

codex/common/model.py

@@ -5,7 +5,7 @@
 from prisma.models import Function, ObjectField, ObjectType
 from pydantic import BaseModel, Field
 
-from codex.api_model import Pagination
+from codex.api_model import ObjectTypeModel, Pagination
 from codex.common.database import INCLUDE_FIELD, INCLUDE_TYPE
 from codex.common.types import (
     extract_field_type,
@@ -15,67 +15,6 @@
 )
 
 
-class ObjectTypeModel(BaseModel):
-    name: str = Field(description="The name of the object")
-    code: Optional[str] = Field(description="The code of the object", default=None)
-    description: Optional[str] = Field(
-        description="The description of the object", default=None
-    )
-    Fields: List["ObjectFieldModel"] = Field(description="The fields of the object")
-    is_pydantic: bool = Field(
-        description="Whether the object is a pydantic model", default=True
-    )
-    is_implemented: bool = Field(
-        description="Whether the object is implemented", default=True
-    )
-    is_enum: bool = Field(description="Whether the object is an enum", default=False)
-
-    def __init__(self, db_obj: ObjectType | None = None, **data):
-        if not db_obj:
-            super().__init__(**data)
-            return
-
-        super().__init__(
-            name=db_obj.name,
-            code=db_obj.code,
-            description=db_obj.description,
-            is_pydantic=db_obj.isPydantic,
-            is_enum=db_obj.isEnum,
-            Fields=[ObjectFieldModel(db_obj=f) for f in db_obj.Fields or []],
-            **data,
-        )
-
-
-class ObjectFieldModel(BaseModel):
-    name: str = Field(description="The name of the field")
-    description: Optional[str] = Field(
-        description="The description of the field", default=None
-    )
-    type: str = Field(
-        description="The type of the field. Can be a string like List[str] or an use any of they related types like list[User]",
-    )
-    value: Optional[str] = Field(description="The value of the field", default=None)
-    related_types: Optional[List[ObjectTypeModel]] = Field(
-        description="The related types of the field", default=[]
-    )
-
-    def __init__(self, db_obj: ObjectField | None = None, **data):
-        if not db_obj:
-            super().__init__(**data)
-            return
-
-        super().__init__(
-            name=db_obj.name,
-            description=db_obj.description,
-            type=db_obj.typeName,
-            value=db_obj.value,
-            related_types=[
-                ObjectTypeModel(db_obj=t) for t in db_obj.RelatedTypes or []
-            ],
-            **data,
-        )
-
-
 class FunctionDef(BaseModel):
     name: str
     arg_types: List[tuple[str, str]]

codex/database.py

@@ -1,5 +1,13 @@
 from prisma.enums import Role
-from prisma.models import Application, CompiledRoute, CompletedApp, Specification, User
+from prisma.models import (
+    APIRouteSpec,
+    Application,
+    CompiledRoute,
+    CompletedApp,
+    Module,
+    Specification,
+    User,
+)
 from prisma.types import CompletedAppCreateInput, UserCreateWithoutRelationsInput
 
 from codex.api_model import (
@@ -301,3 +309,17 @@ async def create_completed_app(
     )
     app = await CompletedApp.prisma().create(data)
     return app
+
+
+async def get_api_route_by_id(ids: Identifiers, api_route_id: str):
+    api_route = await APIRouteSpec.prisma().find_unique_or_raise(
+        where={"id": api_route_id, "Application": {"id": ids.app_id}}
+    )
+    return api_route
+
+
+async def get_module_by_id(ids: Identifiers, module_id: str):
+    module = await Module.prisma().find_unique_or_raise(
+        where={"id": module_id, "Application": {"id": ids.app_id}}
+    )
+    return module