DTO Factory: piccolo implementation

[peterschutt]

[dantownsend]

It would be nice to get this added - maybe we can help in some way.

We recently added support for litestar to Piccolo (replacing starlite), and came across this in our integration tests. For now, we'll version pin to litestar==2.0.0a3.

[peterschutt]

Hey @dantownsend - thanks for reaching out.

I've only implemented support for dataclasses and sqlalchemy so far as PoC. While we are still in alpha/beta I'm keen to make another pass over the internals of the DTO factory ABC before building implementations for all of the remaining libraries that we had support for in v1.

I'm away on a family holiday for the next week or so and will launch into this stuff 2nd week of may.

Any help along the way would be appreciated, am I ok to @ you in here if I have any questions?

[dantownsend]

@peterschutt Cool, happy to help! Thanks for the update.

[Goldziher]

@peterschutt Cool, happy to help! Thanks for the update.

Hi @dantownsend - i added a PR with a DTO factory for Piccolo. Can you perhaps take a look? Additionally, it would be great if you or one of your collaborators would contribute improvements to this. I did not set out to handle nested models and relations, and that would be a great addition.

[dantownsend]

@Goldziher It looks great - thank you! I'll have a play around with it. @sinisaos in case you want to give it a go.

[sinisaos]

@dantownsend Just finished trying out PiccoloDTO and everything is working great (great job @Goldziher). I had a little problem with the partial update and then I found a very good documentation with an example that helped me to get everything work. @Goldziher I see you write the documentation for PiccoloDTO and here is a full working example for PiccoloDTO that you can use

import typing as t

from litestar import Litestar, MediaType, asgi, delete, get, patch, post
from litestar.contrib.jinja import JinjaTemplateEngine
from litestar.contrib.piccolo.dto import PiccoloDTO
from litestar.dto.factory import DTOConfig, DTOData
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_admin.endpoints import create_admin

from home.endpoints import home
from home.piccolo_app import APP_CONFIG
from home.tables import Task


# mounting Piccolo Admin
@asgi("/admin/", is_mount=True)
async def admin(scope: "Scope", receive: "Receive", send: "Send") -> None:
    await create_admin(tables=APP_CONFIG.table_classes)(scope, receive, send)


class PatchDTO(PiccoloDTO[Task]):
    """Don't allow client to set the id, and allow partial updates."""

    config = DTOConfig(exclude={"id"}, partial=True)


@get(
    "/tasks",
    return_dto=PiccoloDTO[Task],
    media_type=MediaType.JSON,
    tags=["Task"],
)
async def tasks() -> t.List[Task]:
    return await Task.select().order_by(Task.id, ascending=False)


@post(
    "/tasks",
    dto=PiccoloDTO[Task],
    return_dto=PiccoloDTO[Task],
    media_type=MediaType.JSON,
    tags=["Task"],
)
async def create_task(data: Task) -> Task:
    await data.save()
    await data.refresh()
    return data


@patch(
    "/tasks/{task_id:int}",
    dto=PatchDTO,
    return_dto=PiccoloDTO[Task],
    media_type=MediaType.JSON,
    tags=["Task"],
)
async def update_task(task_id: int, data: DTOData[Task]) -> Task:
    task = await Task.objects().get(Task.id == task_id)
    if not task:
        raise NotFoundException("Task does not exist")
    result = data.update_instance(task)
    await result.save()
    return result


@delete("/tasks/{task_id:int}", tags=["Task"])
async def delete_task(task_id: int) -> None:
    task = await Task.objects().get(Task.id == task_id)
    if not task:
        raise NotFoundException("Task does not exist")
    await task.remove()


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 = Litestar(
    route_handlers=[
        admin,
        home,
        tasks,
        create_task,
        delete_task,
        update_task,
    ],
    template_config=TemplateConfig(
        directory="home/templates", engine=JinjaTemplateEngine
    ),
    static_files_config=[
        StaticFilesConfig(directories=["static"], path="/static/"),
    ],
    on_startup=[open_database_connection_pool],
    on_shutdown=[close_database_connection_pool],
)

Sorry for long comment.

[Goldziher]

@dantownsend Just finished trying out PiccoloDTO and everything is working great (great job @Goldziher). I had a little problem with the partial update and then I found a very good documentation with an example that helped me to get everything work. @Goldziher I see you write the documentation for PiccoloDTO and here is a full working example for PiccoloDTO that you can use

import typing as t

from litestar import Litestar, MediaType, asgi, delete, get, patch, post
from litestar.contrib.jinja import JinjaTemplateEngine
from litestar.contrib.piccolo.dto import PiccoloDTO
from litestar.dto.factory import DTOConfig, DTOData
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_admin.endpoints import create_admin

from home.endpoints import home
from home.piccolo_app import APP_CONFIG
from home.tables import Task


# mounting Piccolo Admin
@asgi("/admin/", is_mount=True)
async def admin(scope: "Scope", receive: "Receive", send: "Send") -> None:
    await create_admin(tables=APP_CONFIG.table_classes)(scope, receive, send)


class PatchDTO(PiccoloDTO[Task]):
    """Don't allow client to set the id, and allow partial updates."""

    config = DTOConfig(exclude={"id"}, partial=True)


@get(
    "/tasks",
    return_dto=PiccoloDTO[Task],
    media_type=MediaType.JSON,
    tags=["Task"],
)
async def tasks() -> t.List[Task]:
    return await Task.select().order_by(Task.id, ascending=False)


@post(
    "/tasks",
    dto=PiccoloDTO[Task],
    return_dto=PiccoloDTO[Task],
    media_type=MediaType.JSON,
    tags=["Task"],
)
async def create_task(data: Task) -> Task:
    await data.save()
    await data.refresh()
    return data


@patch(
    "/tasks/{task_id:int}",
    dto=PatchDTO,
    return_dto=PiccoloDTO[Task],
    media_type=MediaType.JSON,
    tags=["Task"],
)
async def update_task(task_id: int, data: DTOData[Task]) -> Task:
    task = await Task.objects().get(Task.id == task_id)
    if not task:
        raise NotFoundException("Task does not exist")
    result = data.update_instance(task)
    await result.save()
    return result


@delete("/tasks/{task_id:int}", tags=["Task"])
async def delete_task(task_id: int) -> None:
    task = await Task.objects().get(Task.id == task_id)
    if not task:
        raise NotFoundException("Task does not exist")
    await task.remove()


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 = Litestar(
    route_handlers=[
        admin,
        home,
        tasks,
        create_task,
        delete_task,
        update_task,
    ],
    template_config=TemplateConfig(
        directory="home/templates", engine=JinjaTemplateEngine
    ),
    static_files_config=[
        StaticFilesConfig(directories=["static"], path="/static/"),
    ],
    on_startup=[open_database_connection_pool],
    on_shutdown=[close_database_connection_pool],
)

Sorry for long comment.

cheers, would you care to add a PR to the docs?

[Goldziher]

@dantownsend and @sinisaos - if you guys would like to add a followup PR that improves the DTO by supporting recursive relations etc. that would be great. We have very strong support for SQLA, because we have the knowledge to do this. I do not have the knowledge how to handle Piccolo on this level, but it would be great to have.

[sinisaos]

@Goldziher Sorry for the slow response. I made a PR #1916 for the PiccoloDTO docs and I hope it's good enough. If not, feel free to change it.