Compatibility with Sanic

[georges-g]

Hello,

So far, I am using Sanic with Tortoise-ORM / Aerich, but I'm considering switching my ORM to Piccolo.

Would it work with Sanic though?
If you made it work and have instructions about it, I'm interested.

Best

[dantownsend]

@georges-g Yeah, Piccolo should work fine with Sanic.

The reason we don't have a built-in template for Sanic is because I'm not sure if it allows you to mount ASGI apps within it.

Piccolo Admin is an ASGI application, and we usually mount it at /admin/ within the main application. If this isn't possible with Sanic, it's not the end of the world, because Piccolo Admin can just be run as a separate application.

Usually we create a connection pool when the application starts. Here's an example with FastAPI - I'm sure something similar is possible with Sanic's life cycle hooks:

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

Lines 80 to 95 in 17e9448

	@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")

But again, this is optional. A connection pool isn't required for Piccolo to work.

[sinisaos]

@georges-g This is a simple example of a Sanic application (with Sanic Pydantic) and Piccolo ORM (Sqlite) (based on this project structure).

# content of sanic_app.py
import asyncio
import typing as t

import uvicorn
from piccolo.apps.user.tables import BaseUser
from piccolo.columns import Boolean, ForeignKey, Interval, Varchar
from piccolo.table import Table, create_db_tables
from piccolo.utils.pydantic import create_pydantic_model
from sanic import Sanic, json
from sanic_pydantic import webargs

from piccolo_conf import DB


class Task(Table, db=DB):
    name = Varchar()
    completed = Boolean(default=False)
    task_user = ForeignKey(references=BaseUser)


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",
    nested=True,
)


app = Sanic("PiccoloApp")


@app.get("/tasks/")
async def tasks(request, **kwargs):
    tasks = (
        await Task.select(
            Task.all_columns(),
            Task.task_user.id,
            Task.task_user.username,
        )
        .order_by(Task.id, ascending=False)
        .output(nested=True)
    )
    # return list of Pydantic models in response
    # return json([TaskModelOut(**item).model_dump() for item in tasks])
    # or return json of tasks
    return json(tasks)


@app.get("/tasks/<task_id:int>/")
async def single_task(request, task_id):
    task = (
        await Task.select(
            Task.all_columns(),
            Task.task_user.id,
            Task.task_user.username,
        )
        .where(Task.id == task_id)
        .first()
        .output(nested=True)
    )
    # return Pydantic models in response
    # return json(TaskModelOut(**task).model_dump())
    # or return json of tasks
    return json(task)


@app.post("/tasks/")
@webargs(body=TaskModelIn)
async def create_task(request, **kwargs):
    task = Task(**kwargs["payload"])
    await task.save()
    return json(task.to_dict())


@app.put("/tasks/<task_id:int>/")
@webargs(body=TaskModelIn)
async def update_task(request, task_id, **kwargs):
    task = await Task.objects().get(Task.id == task_id)
    if not task:
        return json({"msg": "Task does not exist"})
    for key, value in kwargs["payload"].items():
        setattr(task, key, value)

    await task.save()
    return json(task.to_dict())


@app.delete("/tasks/<task_id:int>/")
async def delete_task(request, task_id: int):
    task = await Task.objects().get(Task.id == task_id)
    if not task:
        return json({"msg": "Task does not exist"})
    await task.remove()
    return json({})


async def main():
    # Tables creating
    await create_db_tables(
        BaseUser,
        Task,
        if_not_exists=True,
    )

    # Creating user
    if not await BaseUser.exists().where(BaseUser.email == "admin@test.com"):
        user = BaseUser(
            username="piccolo",
            password="piccolo123",
            email="admin@test.com",
            admin=True,
            active=True,
            superuser=True,
        )
        await user.save()


if __name__ == "__main__":
    asyncio.run(main())

    uvicorn.run(app, host="127.0.0.1", port=8000)

Hope that helps and gives you an idea of ​​how to work with Piccolo.

[georges-g]

@dantownsend, @sinisaos Thanks for the answers, it's very helpful and I will definitely consider Piccolo now (I still have to read the docs 😉).

Usually we create a connection pool when the application starts. Here's an example with FastAPI - I'm sure something similar is possible with Sanic's life cycle hooks

Yes, Sanic has several listeners (and signals) that should work fine.

The reason we don't have a built-in template for Sanic is because I'm not sure if it allows you to mount ASGI apps within it.

Piccolo Admin is an ASGI application, and we usually mount it at /admin/ within the main application. If this isn't possible with Sanic, it's not the end of the world, because Piccolo Admin can just be run as a separate application.

If I understand properly, Piccolo ORM should work fine, but you are not sure about Piccolo Admin integrating properly, right?

If you want to investigate this issue further, the Sanic guys are pretty responsive. Otherwise, I could look into that later.

[sinisaos]

Yes, Sanic has several listeners (and signals) that should work fine.

I've never used Sanic before, but listeners are fine for connecting Postgres like this

from piccolo.engine import engine_finder # import engine finder


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")


@app.listener("before_server_start")
async def startup(app):
    await open_database_connection_pool()


@app.listener("before_server_stop")
async def shutdown(app):
    await close_database_connection_pool()

If I understand properly, Piccolo ORM should work fine, but you are not sure about Piccolo Admin integrating properly, right?

Sanic does not have a mounting option like FastAPI or Starlette. I think that the only way Piccolo Admin could work is to mount as external app. Sorry if I'm wrong.