add BlackSheep as an option to piccolo asgi new

Author: dantownsend

piccolo/apps/asgi/commands/new.py

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

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

@@ -0,0 +1,113 @@
+import pdb
+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 openapidocs.v3 import Info
+
+from home.endpoints import home
+from home.tables import Task
+
+
+app = Application()
+
+
+docs = OpenAPIHandler(info=Info(title="Example API", version="0.0.1"))
+docs.bind_app(app)
+
+
+app.serve_files("static", root_path="/static")
+
+
+app.router.add_get("/", home)
+
+
+TaskModelIn = create_pydantic_model(table=Task, model_name="TaskModelIn")
+TaskModelOut = create_pydantic_model(
+    table=Task, include_default_columns=True, model_name="TaskModelOut"
+)
+TaskModelPartial = create_pydantic_model(
+    table=Task, model_name="TaskModelPartial", all_optional=True
+)
+
+
+@app.router.get("/tasks/")
+async def tasks() -> t.List[TaskModelOut]:
+    return await Task.select().order_by(Task.id).run()
+
+
+@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__)
+
+
+@app.router.put("/tasks/{task_id}/")
+async def put_task(
+    task_id: int, task: FromJSON[TaskModelIn]
+) -> TaskModelOut:
+    _task = await Task.objects().where(Task.id == task_id).first().run()
+    if not _task:
+        return json({}, status=404)
+
+    for key, value in task.value.__dict__.items():
+        setattr(_task, key, value)
+
+    await _task.save().run()
+
+    return TaskModelOut(**_task.__dict__)
+
+
+@app.router.patch("/tasks/{task_id}/")
+async def patch_task(
+    task_id: int, task: FromJSON[TaskModelPartial]
+) -> TaskModelOut:
+    _task = await Task.objects().where(Task.id == task_id).first().run()
+    if not _task:
+        return json({}, status=404)
+
+    for key, value in task.value.__dict__.items():
+        if  value is not None:
+            setattr(_task, key, value)
+
+    await _task.save().run()
+
+    return TaskModelOut(**_task.__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()
+    if not task:
+        return json({}, status=404)
+
+    await task.remove().run()
+
+    return json({})
+
+
+async def open_database_connection_pool(application):
+    try:
+        engine = engine_finder()
+        await engine.start_connection_pool()
+    except Exception:
+        print("Unable to connect to the database")
+
+
+async def close_database_connection_pool(application):
+    try:
+        engine = engine_finder()
+        await engine.close_connection_pool()
+    except Exception:
+        print("Unable to connect to the database")
+
+
+app.on_start += open_database_connection_pool
+app.on_stop += close_database_connection_pool

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

@@ -10,7 +10,6 @@ from starlette.staticfiles import StaticFiles
 
 from home.endpoints import HomeEndpoint
 from home.tables import Task
-from home.piccolo_app import APP_CONFIG
 
 
 app = FastAPI(

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

@@ -6,7 +6,6 @@ from starlette.applications import Starlette
 from starlette.staticfiles import StaticFiles
 
 from home.endpoints import HomeEndpoint
-from home.piccolo_app import APP_CONFIG
 from home.tables import Task
 
 

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

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

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

@@ -0,0 +1,21 @@
+import os
+
+from blacksheep.contents import Content
+from blacksheep.messages import Response
+import jinja2
+
+
+ENVIRONMENT = jinja2.Environment(
+    loader=jinja2.FileSystemLoader(
+        searchpath=os.path.join(os.path.dirname(__file__), "templates")
+    )
+)
+
+
+def home():
+    template = ENVIRONMENT.get_template("home.html.jinja")
+    content = template.render(title="Piccolo + ASGI",)
+    return Response(
+        200,
+        content=Content(b"text/html", bytes(content, encoding='utf8'))
+    )

piccolo/apps/asgi/commands/templates/starlette/home/_starlette_endpoints.py.jinja

@@ -0,0 +1,21 @@
+import os
+
+import jinja2
+from starlette.endpoints import HTTPEndpoint
+from starlette.responses import HTMLResponse
+
+
+ENVIRONMENT = jinja2.Environment(
+    loader=jinja2.FileSystemLoader(
+        searchpath=os.path.join(os.path.dirname(__file__), "templates")
+    )
+)
+
+
+class HomeEndpoint(HTTPEndpoint):
+    async def get(self, request):
+        template = ENVIRONMENT.get_template("home.html.jinja")
+
+        content = template.render(title="Piccolo + ASGI",)
+
+        return HTMLResponse(content)

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

@@ -1,21 +1,5 @@
-import os
-
-import jinja2
-from starlette.endpoints import HTTPEndpoint
-from starlette.responses import HTMLResponse
-
-
-ENVIRONMENT = jinja2.Environment(
-    loader=jinja2.FileSystemLoader(
-        searchpath=os.path.join(os.path.dirname(__file__), "templates")
-    )
-)
-
-
-class HomeEndpoint(HTTPEndpoint):
-    async def get(self, request):
-        template = ENVIRONMENT.get_template("home.html.jinja")
-
-        content = template.render(title="Piccolo + ASGI",)
-
-        return HTMLResponse(content)
+{% if router in ['fastapi', 'starlette'] %}
+    {% include '_starlette_endpoints.py.jinja' %}
+{% elif router == 'blacksheep' %}
+    {% include '_blacksheep_endpoints.py.jinja' %}
+{% endif %}

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

@@ -46,6 +46,10 @@
                 <li><a href="/admin/">Admin</a></li>
                 <li><a href="/tasks/">JSON endpoint</a></li>
             </ul>
+            <h3>BlackSheep</h3>
+            <ul>
+                <li><a href="/docs/">Swagger API</a></li>
+            </ul>
         </section>
     </div>
 {% endblock content %}