Parsing uploaded file using DTO as type for msgspec

[alaws-USGS]

Hello, I am working on an endpoint to parse a custom formatted config file into and commit it to a database and admittedly still learning about Litestar. All of the data is read in as strings with a custom parse function read_file, which returns a dictionary of all the data. I then want to commit each row of that parsed data to our database. I am using the SQLAlchemy Plugin and SQLAlchemy models for validation.

We currently have a DTO called InputSiteInfoDTO that has all the fields we expect from the uploaded file. However, we want to have the file contents be parsed from string to the correct type, something that the pydantic.prase_obj() method does. Here is my current implementation that uses a Pydantic model, InputSiteInfoPydantic.

@post(path="/input_site_infos", response_model=List[InputSiteInfoDTO], tags=["Input Site Info"])
    async def insert_input_site_infos(self, input_site_info_service: InputSiteInfoService, data: Annotated[UploadFile, Body(media_type=RequestEncodingType.MULTI_PART)],) -> List[InputSiteInfo]:
        """Insert multiple site info records from a file into the database via file."""
        content = await data.read()
        
        # read_file is a custom function that calls csv.DictReader and returns the data all as strings
        file_rows = read_file(content, delimiter="\t")  # Use TSV delimiter, change to ',' for CSV if needed
        inserted_sites = []

        # Process each row from the file and insert into DB using the service
        for row in file_rows:
            pydantic_data = InputSiteInfoPydantic.parse_obj(row)

            # Convert to the database model and insert into the database
            station_dto = InputSiteInfo(**pydantic_data.dict())
            inserted_site = await input_site_info_service.create(station_dto)  # Insert into DB (simulated)
            inserted_sites.append(input_site_info_service.to_schema(inserted_site))  # Convert to schema and add to result

        # Return the result as a list of inserted sites
        return inserted_sites

Would it be possible to get this coercion using the decode_builtins method of DTOs? Am I try to do too much and will litestar or the input_site_info_service handle this (below is my service definition)?

"""This sets up Advanced Alchemy repositories and services for Input Site Info."""

from collections.abc import AsyncGenerator

import litestar.plugins.sqlalchemy as sa
from flood_frequency_database_db.models import InputSiteInfo as InputSiteInfoModel
from sqlalchemy.ext.asyncio import AsyncSession


class InputSiteInfoRepository(sa.repository.SQLAlchemyAsyncRepository[InputSiteInfoModel]):
    """Input Site Info repository."""

    model_type = InputSiteInfoModel


class InputSiteInfoService(sa.service.SQLAlchemyAsyncRepositoryService[InputSiteInfoModel]):
    """Info Site Info service."""

    repository_type = InputSiteInfoRepository


async def provide_input_site_info_service(db_session: AsyncSession) -> AsyncGenerator[InputSiteInfoService, None]:
    """Dependency injection provider: provides default input site info repo."""
    async with InputSiteInfoService.new(session=db_session) as service:
        yield service

Thanks and I appreciate your eyes on this!

[alaws-USGS]

I've been doing some more digging into msgspec, which I understand is what litestar uses under the hood for serialization/validation for DTOs. I believe what I could do is just replace the pydantic.BaseModel with a msgspec.Struct and that would cut the pydantic dependency.

However, I feel there should be a way to use a created DTO and use something like this to perform the parsing:

# controller path definition
content = data.read()
decoded_content = msgspec.json.decode(content, type=InputSiteInfoDTO)

Am I getting closer to a solution? Thanks!

[provinzkraut]

You should be able to use .decode_builtins():

@post(path="/input_site_infos", response_model=List[InputSiteInfoDTO], tags=["Input Site Info"])
    async def insert_input_site_infos(self, 
        input_site_info_service: InputSiteInfoService, 
        data: Annotated[UploadFile, Body(media_type=RequestEncodingType.MULTI_PART)],
        request: Request
    ) -> List[InputSiteInfo]:
        """Insert multiple site info records from a file into the database via file."""
        content = await data.read()
        
        # read_file is a custom function that calls csv.DictReader and returns the data all as strings
        file_rows = read_file(content, delimiter="\t")  # Use TSV delimiter, change to ',' for CSV if needed
        inserted_sites = []
 

        # bind the DTO to the current request; This is necessary so the DTO backend
        # has access the correct decoders defined for this handler
        bound_dto = InputSiteInfoDTO(request)

        # Process each row from the file and insert into DB using the service
        for row in file_rows:

            # create a DTO instance from the data
            station_dto = bound_dto.decode_builtins(row)

            inserted_site = await input_site_info_service.create(station_dto)  # Insert into DB (simulated)
            inserted_sites.append(input_site_info_service.to_schema(inserted_site))  # Convert to schema and add to result

        # Return the result as a list of inserted sites
        return inserted_sites

[mtleslieUSGS]

Hi, thank you for the code sample. When I try to run the .decode_builtins() function, the bound DTO raises a key error trying to access the handler's ["data_backend"]. I poked around in the code and saw that the ["return_backend"] is being set instead.

Manually setting the ["data_backend"] to point the same place as the ["return_backend"] actually does work as expected, with the sole exception being that the DTO complains about the id field not being set (id, created_at, and updated_at are internal, autogenerated SQLAlchemy fields that we explicitly exclude in the DTO config, but it only complains about the id). This can be fixed by adding the field before decoding and removing the field before insertion to the DB, or by removing the response_model=List[InputSiteInfoDTO] parameter.

The code:

    @post(path="/input_site_infos", tags=["Input Site Info"])
    async def insert_input_site_infos(self, 
        input_site_info_service: InputSiteInfoService, 
        data: Annotated[UploadFile, Body(media_type=RequestEncodingType.MULTI_PART)],
        request: Request
    ) -> List[InputSiteInfo]:
        """Insert multiple site info records from a file into the database via file."""
        content = await data.read()
        
        # read_file is a custom function that calls csv.DictReader and returns the data all as strings
        file_rows = read_file(content, delimiter="\t")  # Use TSV delimiter, change to ',' for CSV if needed
        inserted_sites = []
 
        # bind the DTO to the current request; This is necessary so the DTO backend
        # has access the correct decoders defined for this handler
        bound_dto = InputSiteInfoDTO(request)

        # Manually set the data backend function 
        bound_dto._dto_backends[bound_dto.asgi_connection.route_handler.handler_id]["data_backend"] = bound_dto._dto_backends[bound_dto.asgi_connection.route_handler.handler_id]["return_backend"]

        # create a DTO instance from the data
        # .decode_builtins() expects an array of objects, not a single object
        station_dto = bound_dto.decode_builtins(file_rows)

        # Insert the modified DTOs into the database and convert to the schema
        inserted_site_list = await input_site_info_service.create_many(station_dto)  # Insert into DB (simulated)

        # Convert to schema and add to the result list
        for site in inserted_site_list:
            inserted_sites.append(input_site_info_service.to_schema(site))  # Convert to schema and append

        # Return the list of inserted sites
        return inserted_sites

Obviously messing around with the internals doesn't feel like the best solution here, so I'm wondering if there's something critical that I'm missing when I set up my DTO or route handler. Do you know why the data_backend function is not being set? Thanks for taking a look!