replace deprecated events in asgi templates

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

@@ -1,18 +1,42 @@
 import typing as t
+from contextlib import asynccontextmanager
 
 from fastapi import FastAPI
 from fastapi.responses import JSONResponse
+from piccolo.engine import engine_finder
 from piccolo_admin.endpoints import create_admin
 from piccolo_api.crud.serializers import create_pydantic_model
-from piccolo.engine import engine_finder
-from starlette.routing import Route, Mount
+from starlette.routing import Mount, Route
 from starlette.staticfiles import StaticFiles
 
 from home.endpoints import HomeEndpoint
 from home.piccolo_app import APP_CONFIG
 from home.tables import Task
 
 
+async def open_database_connection_pool():
+    try:
+        engine = engine_finder()
+        await engine.start_connection_pool()
+    except Exception:
+        print("Unable to connect to the database")
+
+
+async def close_database_connection_pool():
+    try:
+        engine = engine_finder()
+        await engine.close_connection_pool()
+    except Exception:
+        print("Unable to connect to the database")
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    await open_database_connection_pool()
+    yield
+    await close_database_connection_pool()
+
+
 app = FastAPI(
     routes=[
         Route("/", HomeEndpoint),
@@ -22,21 +46,20 @@ app = FastAPI(
                 tables=APP_CONFIG.table_classes,
                 # Required when running under HTTPS:
                 # allowed_hosts=['my_site.com']
-            )
+            ),
         ),
         Mount("/static/", StaticFiles(directory="static")),
     ],
+    lifespan=lifespan,
 )
 
 
 TaskModelIn: t.Any = create_pydantic_model(
     table=Task,
-    model_name='TaskModelIn'
+    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"
 )
 
 
@@ -45,16 +68,16 @@ async def tasks():
     return await Task.select().order_by(Task.id)
 
 
-@app.post('/tasks/', response_model=TaskModelOut)
+@app.post("/tasks/", response_model=TaskModelOut)
 async def create_task(task_model: TaskModelIn):
     task = Task(**task_model.dict())
     await task.save()
     return task.to_dict()
 
 
-@app.put('/tasks/{task_id}/', response_model=TaskModelOut)
+@app.put("/tasks/{task_id}/", response_model=TaskModelOut)
 async def update_task(task_id: int, task_model: TaskModelIn):
-    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)
 
@@ -66,30 +89,12 @@ async def update_task(task_id: int, task_model: TaskModelIn):
     return task.to_dict()
 
 
-@app.delete('/tasks/{task_id}/')
+@app.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 JSONResponse({}, status_code=404)
 
     await task.remove()
 
     return JSONResponse({})
-
-
-@app.on_event("startup")
-async def open_database_connection_pool():
-    try:
-        engine = engine_finder()
-        await engine.start_connection_pool()
-    except Exception:
-        print("Unable to connect to the database")
-
-
-@app.on_event("shutdown")
-async def close_database_connection_pool():
-    try:
-        engine = engine_finder()
-        await engine.close_connection_pool()
-    except Exception:
-        print("Unable to connect to the database")

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

@@ -1,33 +1,17 @@
+from contextlib import asynccontextmanager
+
+from piccolo.engine import engine_finder
 from piccolo_admin.endpoints import create_admin
 from piccolo_api.crud.endpoints import PiccoloCRUD
-from piccolo.engine import engine_finder
-from starlette.routing import Route, Mount
 from starlette.applications import Starlette
+from starlette.routing import Mount, Route
 from starlette.staticfiles import StaticFiles
 
 from home.endpoints import HomeEndpoint
 from home.piccolo_app import APP_CONFIG
 from home.tables import Task
 
 
-app = Starlette(
-    routes=[
-        Route("/", HomeEndpoint),
-        Mount(
-            "/admin/",
-            create_admin(
-                tables=APP_CONFIG.table_classes,
-                # Required when running under HTTPS:
-                # allowed_hosts=['my_site.com']
-            )
-        ),
-        Mount("/static/", StaticFiles(directory="static")),
-        Mount("/tasks/", PiccoloCRUD(table=Task))
-    ],
-)
-
-
-@app.on_event("startup")
 async def open_database_connection_pool():
     try:
         engine = engine_finder()
@@ -36,10 +20,34 @@ async def open_database_connection_pool():
         print("Unable to connect to the database")
 
 
-@app.on_event("shutdown")
 async def close_database_connection_pool():
     try:
         engine = engine_finder()
         await engine.close_connection_pool()
     except Exception:
         print("Unable to connect to the database")
+
+
+@asynccontextmanager
+async def lifespan(app: Starlette):
+    await open_database_connection_pool()
+    yield
+    await close_database_connection_pool()
+
+
+app = Starlette(
+    routes=[
+        Route("/", HomeEndpoint),
+        Mount(
+            "/admin/",
+            create_admin(
+                tables=APP_CONFIG.table_classes,
+                # Required when running under HTTPS:
+                # allowed_hosts=['my_site.com']
+            ),
+        ),
+        Mount("/static/", StaticFiles(directory="static")),
+        Mount("/tasks/", PiccoloCRUD(table=Task)),
+    ],
+    lifespan=lifespan,
+)