What's the best way to access a database from a worker in regards to a session?

[zmarffy]

This FastAPI project generator generates a DAO for a dummy model with a useful FastAPI dependency of a DB session as an instance variable.

def __init__(self, session: AsyncSession = Depends(get_db_session)):
    self.session = session

I am struggling to understand how to pass this over to a taskiq task so the task can write stuff to my app's database, as it (of course) is not serializable.

[s3rius]

Just pass it as a dependency. Take a look at this docs page. If you used fastapi-template, then integration should be enabled, you just have to use TaskiqDepends for all tasks and don't forget to specify TaskiqDepends as a default value for all depencies that depend on Request or FastAPI.

[zmarffy]

I can't seem to get this to work properly. I have done some extensive research on what the problem is, and I have found the same concept, but in terms of FastAPI's Depends object. Basically, my function is only ever returning a Dependency object, not the actual AsyncSession, so I think I am not passing it down the line correctly. But I followed the examples to a tee so I am not sure what is going on here.

# tasks.py

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from taskiq import TaskiqDepends

from myapp.db.dependencies import get_db_session
from myapp.db.models.user_model import UserModel
from myapp.tasks.tkq import broker

@broker.task
async def data_grab(
    email: str,
    session: AsyncSession = TaskiqDepends(get_db_session),
) -> None:
    rows = await session.execute(
        select(UserModel).where(UserModel.email == email)
    )
    # do stuff with rows here

# dependencies.py

from typing import AsyncGenerator

from sqlalchemy.ext.asyncio import AsyncSession
from starlette.requests import Request
from taskiq import TaskiqDepends


async def get_db_session(
    request: Request = TaskiqDepends(),
) -> AsyncGenerator[AsyncSession, None]:
    session: AsyncSession = request.app.state.db_session_factory()

    try:  # noqa: WPS501
        yield session
    finally:
        await session.commit()
        await session.close()

# views.py

from fastapi import APIRouter

from myapp.tasks.tasks import data_grab


router = APIRouter()


@router.post(
    "/",
    summary="Run a job",
)
async def create_job_model(
    email: str,
) -> None:
    await data_grab.kiq(email)

Upon sending a POST request, this causes the error: AttributeError: 'Dependency' object has no attribute 'execute'.

[s3rius]

Are you sure that you that you specified all tasks files? That might be a problem.

Take a look at --fs-discover option described here. It should make it automatically.

[s3rius]

Anyway. Try updating to https://github.com/taskiq-python/taskiq/releases/tag/0.9.2. This PR should fix these issues.

[zmarffy]

Worked perfectly, thanks! Great project you have here.