Download file from SQLA table

download_mvp.py

import os
import tempfile
from http import HTTPStatus
from typing import Annotated

import sqlalchemy as sa
from fastapi import APIRouter, BackgroundTasks, Depends, FastAPI, HTTPException
from fastapi.responses import FileResponse
from sqlalchemy import select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, undefer


class Resource(DeclarativeBase):
    __tablename__ = "resources"

    id: Mapped[int] = mapped_column(sa.Integer, primary_key=True)
    file_data: Mapped[bytes] = mapped_column(
        sa.LargeBinary, nullable=False, deferred=True
    )


def get_session(): ...


DBSession = Annotated[Session, Depends(get_session)]


router = APIRouter()


@router.get(
    "/{id}",
    status_code=HTTPStatus.OK,
)
def download_data(
    id: int,
    db_session: DBSession,
    background_tasks: BackgroundTasks,
):
    query = (
        select(Resource).where(Resource.id == id).options(undefer(Resource.file_data))
    )
    resource = db_session.scalar(query)
    if resource is None:
        raise HTTPException(
            status_code=HTTPStatus.NOT_FOUND, detail="Resource not found"
        )

    with tempfile.NamedTemporaryFile("w+b", delete=False) as named_temp_file:
        named_temp_file.write(resource.file_data)
        named_temp_file.seek(0)
        background_tasks.add_task(os.remove, named_temp_file.name)
        return FileResponse(
            named_temp_file.name,
            media_type="application/octet-stream",
            filename=f"resource_{id}.extention",
            background=background_tasks,
        )


app = FastAPI()
app.include_router(router)