Update asgi templates

piccolo/apps/asgi/commands/templates/app/_blacksheep_app.py.jinja

@@ -1,20 +1,18 @@
 import typing as t
 
-from piccolo_admin.endpoints import create_admin
-from piccolo_api.crud.serializers import create_pydantic_model
-from piccolo.engine import engine_finder
-
 from blacksheep.server import Application
 from blacksheep.server.bindings import FromJSON
-from blacksheep.server.responses import json
 from blacksheep.server.openapi.v3 import OpenAPIHandler
+from blacksheep.server.responses import json
 from openapidocs.v3 import Info
+from piccolo.engine import engine_finder
+from piccolo_admin.endpoints import create_admin
+from piccolo_api.crud.serializers import create_pydantic_model
 
 from home.endpoints import home
 from home.piccolo_app import APP_CONFIG
 from home.tables import Task
 
-
 app = Application()
 
 app.mount(
@@ -47,7 +45,7 @@ TaskModelPartial: t.Any = create_pydantic_model(
 
 @app.router.get("/tasks/")
 async def tasks() -> t.List[TaskModelOut]:
-    return await Task.select().order_by(Task.id)
+    return await Task.select().order_by(Task._meta.primary_key, ascending=False)
 
 
 @app.router.post("/tasks/")
@@ -58,10 +56,8 @@ async def create_task(task_model: FromJSON[TaskModelIn]) -> TaskModelOut:
 
 
 @app.router.put("/tasks/{task_id}/")
-async def put_task(
-    task_id: int, task_model: FromJSON[TaskModelIn]
-) -> TaskModelOut:
-    task = await Task.objects().get(Task.id == task_id)
+async def put_task(task_id: int, task_model: FromJSON[TaskModelIn]) -> TaskModelOut:
+    task = await Task.objects().get(Task._meta.primary_key == task_id)
     if not task:
         return json({}, status=404)
 
@@ -77,7 +73,7 @@ async def put_task(
 async def patch_task(
     task_id: int, task_model: FromJSON[TaskModelPartial]
 ) -> TaskModelOut:
-    task = await Task.objects().get(Task.id == task_id)
+    task = await Task.objects().get(Task._meta.primary_key == task_id)
     if not task:
         return json({}, status=404)
 
@@ -92,7 +88,7 @@ async def patch_task(
 
 @app.router.delete("/tasks/{task_id}/")
 async def delete_task(task_id: int):
-    task = await Task.objects().get(Task.id == task_id)
+    task = await Task.objects().get(Task._meta.primary_key == task_id)
     if not task:
         return json({}, status=404)
 

piccolo/apps/asgi/commands/templates/app/_esmerald_app.py.jinja

@@ -1,23 +1,21 @@
 import typing as t
-
 from pathlib import Path
 
-from piccolo.utils.pydantic import create_pydantic_model
-from piccolo.engine import engine_finder
-from piccolo_admin.endpoints import create_admin
-
 from esmerald import (
+    APIView,
     Esmerald,
-    Include,
     Gateway,
+    Include,
     JSONResponse,
-    APIView,
+    delete,
     get,
     post,
     put,
-    delete
 )
 from esmerald.config import StaticFilesConfig
+from piccolo.engine import engine_finder
+from piccolo.utils.pydantic import create_pydantic_model
+from piccolo_admin.endpoints import create_admin
 
 from home.endpoints import home
 from home.piccolo_app import APP_CONFIG
@@ -40,14 +38,9 @@ async def close_database_connection_pool():
         print("Unable to connect to the database")
 
 
-TaskModelIn: t.Any = create_pydantic_model(
-    table=Task,
-    model_name='TaskModelIn'
-)
+TaskModelIn: t.Any = create_pydantic_model(table=Task, model_name="TaskModelIn")
 TaskModelOut: t.Any = create_pydantic_model(
-    table=Task,
-    include_default_columns=True,
-    model_name='TaskModelOut'
+    table=Task, include_default_columns=True, model_name="TaskModelOut"
 )
 
 
@@ -57,19 +50,17 @@ class TaskAPIView(APIView):
 
     @get("/")
     async def tasks(self) -> t.List[TaskModelOut]:
-        return await Task.select().order_by(Task.id)
-
+        return await Task.select().order_by(Task._meta.primary_key, ascending=False)
 
-    @post('/')
+    @post("/")
     async def create_task(self, payload: TaskModelIn) -> TaskModelOut:
         task = Task(**payload.dict())
         await task.save()
         return task.to_dict()
 
-
-    @put('/{task_id}')
+    @put("/{task_id}")
     async def update_task(self, payload: TaskModelIn, task_id: int) -> TaskModelOut:
-        task = await Task.objects().get(Task.id == task_id)
+        task = await Task.objects().get(Task._meta.primary_key == task_id)
         if not task:
             return JSONResponse({}, status_code=404)
 
@@ -80,10 +71,9 @@ class TaskAPIView(APIView):
 
         return task.to_dict()
 
-
-    @delete('/{task_id}')
+    @delete("/{task_id}")
     async def delete_task(self, task_id: int) -> None:
-        task = await Task.objects().get(Task.id == task_id)
+        task = await Task.objects().get(Task._meta.primary_key == task_id)
         if not task:
             return JSONResponse({}, status_code=404)
 

piccolo/apps/asgi/commands/templates/app/_lilya_app.py.jinja

@@ -1,15 +1,14 @@
-from piccolo_admin.endpoints import create_admin
-from piccolo_api.crud.endpoints import PiccoloCRUD
-from piccolo.engine import engine_finder
-from lilya.routing import Path, Include
 from lilya.apps import Lilya
+from lilya.routing import Include, Path
 from lilya.staticfiles import StaticFiles
+from piccolo.engine import engine_finder
+from piccolo_admin.endpoints import create_admin
+from piccolo_api.crud.endpoints import PiccoloCRUD
 
 from home.endpoints import HomeController
 from home.piccolo_app import APP_CONFIG
 from home.tables import Task
 
-
 app = Lilya(
     routes=[
         Path("/", HomeController),
@@ -19,10 +18,10 @@ app = Lilya(
                 tables=APP_CONFIG.table_classes,
                 # Required when running under HTTPS:
                 # allowed_hosts=['my_site.com']
-            )
+            ),
         ),
         Include("/static/", StaticFiles(directory="static")),
-        Include("/tasks/", PiccoloCRUD(table=Task))
+        Include("/tasks/", PiccoloCRUD(table=Task)),
     ],
 )
 

piccolo/apps/asgi/commands/templates/app/_litestar_app.py.jinja

@@ -1,17 +1,25 @@
 import typing as t
 
-from home.endpoints import home
-from home.piccolo_app import APP_CONFIG
-from home.tables import Task
 from litestar import Litestar, asgi, delete, get, patch, post
 from litestar.contrib.jinja import JinjaTemplateEngine
 from litestar.exceptions import NotFoundException
 from litestar.static_files import StaticFilesConfig
 from litestar.template import TemplateConfig
 from litestar.types import Receive, Scope, Send
 from piccolo.engine import engine_finder
-from piccolo.utils.pydantic import create_pydantic_model
 from piccolo_admin.endpoints import create_admin
+from pydantic import BaseModel
+
+from home.endpoints import home
+from home.piccolo_app import APP_CONFIG
+from home.tables import Task
+
+"""
+NOTE: `create_pydantic_model` is not compatible with Litestar
+version higher than 2.11.0. If you are using Litestar<=2.11.0, 
+you can use `create_pydantic_model` as in other asgi templates
+
+from piccolo.utils.pydantic import create_pydantic_model
 
 TaskModelIn: t.Any = create_pydantic_model(
     table=Task,
@@ -22,6 +30,18 @@ TaskModelOut: t.Any = create_pydantic_model(
     include_default_columns=True,
     model_name="TaskModelOut",
 )
+"""
+
+
+class TaskModelIn(BaseModel):
+    name: str
+    completed: bool = False
+
+
+class TaskModelOut(BaseModel):
+    id: int
+    name: str
+    completed: bool = False
 
 
 # mounting Piccolo Admin
@@ -32,31 +52,32 @@ async def admin(scope: "Scope", receive: "Receive", send: "Send") -> None:
 
 @get("/tasks", tags=["Task"])
 async def tasks() -> t.List[TaskModelOut]:
-    return await Task.select().order_by(Task.id, ascending=False)
+    tasks = await Task.select().order_by(Task._meta.primary_key, ascending=False)
+    return [TaskModelOut(**task) for task in tasks]
 
 
 @post("/tasks", tags=["Task"])
 async def create_task(data: TaskModelIn) -> TaskModelOut:
-    task = Task(**data.dict())
+    task = Task(**data.model_dump())
     await task.save()
-    return task.to_dict()
+    return TaskModelOut(**task.to_dict())
 
 
 @patch("/tasks/{task_id:int}", tags=["Task"])
 async def update_task(task_id: int, data: TaskModelIn) -> TaskModelOut:
-    task = await Task.objects().get(Task.id == task_id)
+    task = await Task.objects().get(Task._meta.primary_key == task_id)
     if not task:
         raise NotFoundException("Task does not exist")
-    for key, value in data.dict().items():
+    for key, value in data.model_dump().items():
         setattr(task, key, value)
 
     await task.save()
-    return task.to_dict()
+    return TaskModelOut(**task.to_dict())
 
 
 @delete("/tasks/{task_id:int}", tags=["Task"])
 async def delete_task(task_id: int) -> None:
-    task = await Task.objects().get(Task.id == task_id)
+    task = await Task.objects().get(Task._meta.primary_key == task_id)
     if not task:
         raise NotFoundException("Task does not exist")
     await task.remove()