Accessing M2M models in the handler

[Daniel-Sedlacek]

Hello everyone, first of all I want to give a big thumbs up for Litestar, it's a great project and I'm truly happy to be able to use it.
However, I got into this problem that I can't solve.

The Problem

I'm struggling with retrieving nested data from many-to-many relationship model in a route handler. Specifically, I need to handle nested service data when creating appointments.

Code Example

Here's my model structure that uses the M2M implementation according to Piccolo's docs:

class Service(Table):
    id = UUID(primary_key=True)
    name = Varchar(length=100, null=False)
    appointments = M2M(LazyTableReference("AppointmentService", module_path=__name__))


class Appointment(Table):
    id = UUID(primary_key=True)
    customer_name = Varchar(length=100, null=True)
    services = M2M(LazyTableReference("AppointmentService", module_path=__name__))


class AppointmentToService(Table):
   id = UUID(primary_key=True)
    appointment = ForeignKey(references=Appointment)
    service = ForeignKey(references=Service)

When I send this JSON payload from the frontend:

appointment_json_payload = {
    "customer_name": "John Doe",
    "services": [{"id": "123e4567-e89b-12d3-a456-426614174000", "name": "Haircut"}],
}

My current DTOs look like this:

class ServiceDTO(PiccoloDTO[Service]):
    config = DTOConfig()


class AppointmentCreateDTO(PiccoloDTO[Appointment]):
    services: List[ServiceDTO]
    config = DTOConfig(exclude={"id"}, partial=True)

And the route handler:

class AppointmentController(Controller):
    @post("/appointments", dto=AppointmentCreateDTO, return_dto=AppointmentCreateDTO)
    async def create_appointment(
        self,
        data: DTOData[Appointment],
    ) -> Appointment:
        # Create appointment instance
        appointment_data = data.create_instance()

        return appointment_data

Question

How can I properly parse this nested data structure using DTOs and then retrieve the whole Appointment model, including the nested Service data? The data.create_instance() doesn't include the nested service data in this case.

What I expect to get from data.create_instance():

Appointment(
    customer_name="John Doe",
    services=[
        Service(
            id="123e4567-e89b-12d3-a456-426614174000",
            name="Haircut"
        )
    ]
)

but what I actually get:

Appointment(
    customer_name="John Doe",
    services=[]  # Empty list, nested services data is lost
)

Any help would be greatly appreciated! 🙏

[sinisaos]

@Daniel-Sedlacek PiccoloDTO cannot handle nested data, but luckily Piccolo has an output method that can handle deeply nested data and you can use that. You can use a Pydantic model with an output(nestad=True) method like this.

import typing as t
from pydantic import BaseModel, UUID4

class ServiceModelIn(BaseModel):
    name: str
    appointments: t.List[UUID4]


class ServiceModelOut(BaseModel):
    id: UUID4
    name: str


class AppointmentModelIn(BaseModel):
    customer_name: str
    services: t.List[UUID4]


class AppointmentModelOut(BaseModel):
    id: UUID4
    customer_name: str
    services: t.List[ServiceModelOut]


@get("/appointments", tags=["Appointment"])
async def appointments() -> t.List[AppointmentModelOut]:
    results = (
        await Appointment.select(
            Appointment.all_columns(),
            Appointment.services(),
        )
        .order_by(Appointment._meta.primary_key, ascending=False)
        .output(nested=True)
    )
    # print([AppointmentModelOut(**result) for result in results])
    #[
    #     AppointmentModelOut(
    #         id=UUID("c67439d0-d1e0-4061-98f3-d40068157f0a"),
    #         customer_name="John Doe",
    #         services=[
    #             ServiceModelOut(
    #                 id=UUID("c2ed749b-345c-4bdd-8dd3-2f65ddd472f3"), name="Haircut"
    #             )
    #         ],
    #     )
    # ]
    return [AppointmentModelOut(**result) for result in results]

Use the same code if you want to return a nested response from the post method. If you need an example of the post method, I can provide it in future comments. Hope this helps.

[Daniel-Sedlacek]

Thanks a lot, that's a great insight! I will definitely use it in my work.
However, I'm unsure if this solves my problem in receiving and parsing data in a POST handler?
I will post my solution, even though I'm not sure if this is the most elegant way.

1. I parse the data into the Appointment model instance using the PiccoloDTO and data.create_instance() method
2. I receive and validate the Services from the request body using msgspec and later manually handle the M2M relationship (not shown)

from msgspec import Struct
from msgspec.json import decode

class AppointmentServiceValidation(Struct):
    service: str  # UUID of the service
    price_at_booking: Optional[float]
    duration_at_booking: Optional[int]

class AppointmentServicesPayload(Struct):
    services: List[AppointmentServiceValidation]

 @post("/appointments", dto=AppointmentCreateDTO, return_dto=AppointmentDTO)
    async def create_appointment(
        self,
        request: Request[User, Token, Any],
        data: DTOData[Appointment],
    ) -> Appointment:

# Create appointment instance
appointment_data = data.create_instance()

# Get and validate services data from the request
request_body = await request.body()  # Get raw bytes
validated_payload = decode(request_body, type=AppointmentServicesPayload)
services_data = validated_payload.services

[sinisaos]

@Daniel-Sedlacek I have never used msgspec but I think it is a similar library like Pydantic (for parsing and validating data). When I need to save M2M record to database with Piccolo (I usually use FastAPI with Pydantic and Piccolo as query builder). Try something like this.

# use Pydantic models from the previous example

@post("/appointments", tags=["Appointment"])
async def create_appointment(data: AppointmentModelIn) -> AppointmentModelOut:
    # create new record and return id
    new_appointment = await Appointment.insert(
        Appointment(customer_name=data.customer_name)
    ).returning(Appointment._meta.primary_key)
    # if the M2M record does not exist raise exception
    try:
        for service in data.services:
            # create M2M record through joining table
            appointment_service = AppointmentToService(
                appointment=new_appointment[0]["id"],
                service=service,
            )
            # save M2M record
            await appointment_service.save()
    except Exception:
        # delete new_appointment if FK violation raised
        await Appointment.delete().where(
            Appointment._meta.primary_key == new_appointment[0]["id"]
        )
        raise HTTPException(
            detail="Foreign key constraint violation. Record is not present in M2M table.",
            status_code=422,
        )
    # return nested result
    return (
        await Appointment.select(
            Appointment.all_columns(),
            Appointment.services(),
        )
        .where(Appointment._meta.primary_key == new_appointment[0]["id"])
        .first()
        .output(nested=True)
    )

In Piccolo I mostly use the query builder, but if you want to use an ORM with objects, change example to suit your needs.

[Daniel-Sedlacek]

@sinisaos Thanks for the reply, I will let you know how here as soon as I try it with my code. Nevertheless, I appreciate your time writing these answers! :-) Cheers from Austria