asgi template for Xpresso framework (#415)

* asgi template for Xpresso framework

* add lifespan context managers

* apply suggestions and update readme.md

* add xpresso and blacksheep to integration test, and modernise older ASGI templates

Co-authored-by: Daniel Townsend <[email protected]>

README.md

@@ -96,7 +96,7 @@ Let Piccolo scaffold you an ASGI web app, using Piccolo as the ORM:
 piccolo asgi new
 ```
 
-[Starlette](https://www.starlette.io/), [FastAPI](https://fastapi.tiangolo.com/), and [BlackSheep](https://www.neoteroi.dev/blacksheep/) are currently supported.
+[Starlette](https://www.starlette.io/), [FastAPI](https://fastapi.tiangolo.com/), [BlackSheep](https://www.neoteroi.dev/blacksheep/) and [Xpresso](https://xpresso-api.dev/) are currently supported.
 
 ## Are you a Django user?
 

docs/src/piccolo/asgi/index.rst

@@ -21,7 +21,8 @@ Routing frameworks
 ******************
 
 Currently, `Starlette <https://www.starlette.io/>`_, `FastAPI <https://fastapi.tiangolo.com/>`_,
-and `BlackSheep <https://www.neoteroi.dev/blacksheep/>`_ are supported.
+`BlackSheep <https://www.neoteroi.dev/blacksheep/>`_ and `Xpresso <https://xpresso-api.dev/>`_
+are supported.
 
 Other great ASGI routing frameworks exist, and may be supported in the future
 (`Quart <https://pgjones.gitlab.io/quart/>`_ ,

piccolo/apps/asgi/commands/new.py

@@ -10,7 +10,7 @@
 
 TEMPLATE_DIR = os.path.join(os.path.dirname(__file__), "templates/app/")
 SERVERS = ["uvicorn", "Hypercorn"]
-ROUTERS = ["starlette", "fastapi", "blacksheep"]
+ROUTERS = ["starlette", "fastapi", "blacksheep", "xpresso"]
 
 
 def print_instruction(message: str):

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

@@ -47,56 +47,56 @@ TaskModelPartial: t.Any = create_pydantic_model(
 
 @app.router.get("/tasks/")
 async def tasks() -> t.List[TaskModelOut]:
-    return await Task.select().order_by(Task.id).run()
+    return await Task.select().order_by(Task.id)
 
 
 @app.router.post("/tasks/")
-async def create_task(task: FromJSON[TaskModelIn]) -> TaskModelOut:
-    task = Task(**task.value.__dict__)
-    await task.save().run()
-    return TaskModelOut(**task.__dict__)
+async def create_task(task_model: FromJSON[TaskModelIn]) -> TaskModelOut:
+    task = Task(**task_model.value.dict())
+    await task.save()
+    return TaskModelOut(**task.to_dict())
 
 
 @app.router.put("/tasks/{task_id}/")
 async def put_task(
-    task_id: int, task: FromJSON[TaskModelIn]
+    task_id: int, task_model: FromJSON[TaskModelIn]
 ) -> TaskModelOut:
-    _task = await Task.objects().where(Task.id == task_id).first().run()
-    if not _task:
+    task = await Task.objects().get(Task.id == task_id)
+    if not task:
         return json({}, status=404)
 
-    for key, value in task.value.__dict__.items():
-        setattr(_task, key, value)
+    for key, value in task_model.value.dict().items():
+        setattr(task, key, value)
 
-    await _task.save().run()
+    await task.save()
 
-    return TaskModelOut(**_task.__dict__)
+    return TaskModelOut(**task.to_dict())
 
 
 @app.router.patch("/tasks/{task_id}/")
 async def patch_task(
-    task_id: int, task: FromJSON[TaskModelPartial]
+    task_id: int, task_model: FromJSON[TaskModelPartial]
 ) -> TaskModelOut:
-    _task = await Task.objects().where(Task.id == task_id).first().run()
-    if not _task:
+    task = await Task.objects().get(Task.id == task_id)
+    if not task:
         return json({}, status=404)
 
-    for key, value in task.value.__dict__.items():
-        if  value is not None:
-            setattr(_task, key, value)
+    for key, value in task_model.value.dict().items():
+        if value is not None:
+            setattr(task, key, value)
 
-    await _task.save().run()
+    await task.save()
 
-    return TaskModelOut(**_task.__dict__)
+    return TaskModelOut(**task.to_dict())
 
 
 @app.router.delete("/tasks/{task_id}/")
 async def delete_task(task_id: int):
-    task = await Task.objects().where(Task.id == task_id).first().run()
+    task = await Task.objects().get(Task.id == task_id)
     if not task:
         return json({}, status=404)
 
-    await task.remove().run()
+    await task.remove()
 
     return json({})
 

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

@@ -1,11 +1,11 @@
 import typing as t
 
+from fastapi import FastAPI
+from fastapi.responses import JSONResponse
 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 fastapi import FastAPI
-from fastapi.responses import JSONResponse
 from starlette.staticfiles import StaticFiles
 
 from home.endpoints import HomeEndpoint
@@ -42,37 +42,37 @@ TaskModelOut: t.Any = create_pydantic_model(
 
 @app.get("/tasks/", response_model=t.List[TaskModelOut])
 async def tasks():
-    return await Task.select().order_by(Task.id).run()
+    return await Task.select().order_by(Task.id)
 
 
 @app.post('/tasks/', response_model=TaskModelOut)
 async def create_task(task_model: TaskModelIn):
-    task = Task(**task_model.__dict__)
-    await task.save().run()
+    task = Task(**task_model.dict())
+    await task.save()
     return task.to_dict()
 
 
 @app.put('/tasks/{task_id}/', response_model=TaskModelOut)
 async def update_task(task_id: int, task_model: TaskModelIn):
-    task = await Task.objects().where(Task.id == task_id).first().run()
+    task = await Task.objects().get(Task.id == task_id)
     if not task:
         return JSONResponse({}, status_code=404)
 
-    for key, value in task_model.__dict__.items():
+    for key, value in task_model.dict().items():
         setattr(task, key, value)
 
-    await task.save().run()
+    await task.save()
 
     return task.to_dict()
 
 
 @app.delete('/tasks/{task_id}/')
 async def delete_task(task_id: int):
-    task = await Task.objects().where(Task.id == task_id).first().run()
+    task = await Task.objects().get(Task.id == task_id)
     if not task:
         return JSONResponse({}, status_code=404)
 
-    await task.remove().run()
+    await task.remove()
 
     return JSONResponse({})
 

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

@@ -0,0 +1,113 @@
+import typing as t
+from contextlib import asynccontextmanager
+
+from piccolo.engine import engine_finder
+from piccolo.utils.pydantic import create_pydantic_model
+from piccolo_admin.endpoints import create_admin
+from starlette.staticfiles import StaticFiles
+from xpresso import App, FromJson, FromPath, HTTPException, Operation, Path
+from xpresso.routing.mount import Mount
+
+from home.endpoints import home
+from home.piccolo_app import APP_CONFIG
+from home.tables import Task
+
+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"
+)
+
+
+async def tasks() -> t.List[TaskModelOut]:
+    return await Task.select().order_by(Task.id)
+
+
+async def create_task(task_model: FromJson[TaskModelIn]) -> TaskModelOut:
+    task = Task(**task_model.dict())
+    await task.save()
+    return task.to_dict()
+
+
+async def update_task(
+    task_id: FromPath[int], task_model: FromJson[TaskModelIn]
+) -> TaskModelOut:
+    task = await Task.objects().get(Task.id == task_id)
+    if not task:
+        raise HTTPException(status_code=404)
+
+    for key, value in task_model.dict().items():
+        setattr(task, key, value)
+
+    await task.save()
+
+    return task.to_dict()
+
+
+async def delete_task(task_id: FromPath[int]):
+    task = await Task.objects().get(Task.id == task_id)
+    if not task:
+        raise HTTPException(status_code=404)
+
+    await task.remove()
+
+    return {}
+
+
+@asynccontextmanager
+async def lifespan():
+    await open_database_connection_pool()
+    try:
+        yield
+    finally:
+        await close_database_connection_pool()
+
+
+app = App(
+    routes=[
+        Path(
+            "/",
+            get=Operation(
+                home,
+                include_in_schema=False,
+            ),
+        ),
+        Mount(
+            "/admin/",
+            create_admin(
+                tables=APP_CONFIG.table_classes,
+                # Required when running under HTTPS:
+                # allowed_hosts=['my_site.com']
+            ),
+        ),
+        Path(
+            "/tasks/",
+            get=tasks,
+            post=create_task,
+            tags=["Task"],
+        ),
+        Path(
+            "/tasks/{task_id}/",
+            put=update_task,
+            delete=delete_task,
+            tags=["Task"],
+        ),
+        Mount("/static/", StaticFiles(directory="static")),
+    ],
+    lifespan=lifespan,
+)
+
+
+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")

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

@@ -4,4 +4,6 @@
     {% include '_starlette_app.py.jinja' %}
 {% elif router == 'blacksheep' %}
     {% include '_blacksheep_app.py.jinja' %}
+{% elif router == 'xpresso' %}
+    {% include '_xpresso_app.py.jinja' %}
 {% endif %}

piccolo/apps/asgi/commands/templates/app/home/_blacksheep_endpoints.py.jinja

@@ -12,7 +12,7 @@ ENVIRONMENT = jinja2.Environment(
 )
 
 
-def home():
+async def home(request):
     template = ENVIRONMENT.get_template("home.html.jinja")
     content = template.render(title="Piccolo + ASGI",)
     return Response(

piccolo/apps/asgi/commands/templates/app/home/_xpresso_endpoints.py.jinja

@@ -0,0 +1,20 @@
+import os
+
+import jinja2
+from xpresso.responses import HTMLResponse
+
+ENVIRONMENT = jinja2.Environment(
+    loader=jinja2.FileSystemLoader(
+        searchpath=os.path.join(os.path.dirname(__file__), "templates")
+    )
+)
+
+
+async def home():
+    template = ENVIRONMENT.get_template("home.html.jinja")
+
+    content = template.render(
+        title="Piccolo + ASGI",
+    )
+
+    return HTMLResponse(content)

piccolo/apps/asgi/commands/templates/app/home/endpoints.py.jinja

@@ -2,4 +2,6 @@
     {% include '_starlette_endpoints.py.jinja' %}
 {% elif router == 'blacksheep' %}
     {% include '_blacksheep_endpoints.py.jinja' %}
+{% elif router == 'xpresso' %}
+    {% include '_xpresso_endpoints.py.jinja' %}
 {% endif %}

piccolo/apps/asgi/commands/templates/app/home/templates/home.html.jinja_raw

@@ -51,6 +51,11 @@
                 <li><a href="/admin/">Admin</a></li>
                 <li><a href="/docs/">Swagger API</a></li>
             </ul>
+            <h3>Xpresso</h3>
+            <ul>
+                <li><a href="/admin/">Admin</a></li>
+                <li><a href="/docs/">Swagger API</a></li>
+            </ul>
         </section>
     </div>
 {% endblock content %}