How to catch exceptions in generator

[vincentleeuwen]

Hey there,

Congrats on this nice library! I really like the idea and am thinking of using it in our project. I have a quick question (not sure if this is the right place to ask, but I'll give it a try).

We want to use this library to resolve fastapi Depends within a celery task (seems like this would be a good fit).

We have the following kind setup:

def get_session() -> Generator[Session, None, None]:
    session = get_session_maker()()
    try:
        yield session
    except Exception:
        session.rollback()
        raise
    else:
        session.commit()
    finally:
        session.close()

Then in the task:

@celery_app.task  # type: ignore[misc]
def run_demo_import_task(file: io.BytesIO) -> None:
    try:
        get_some_service().some_task(file=file)
    except Exception:
        # I am mot sure how to make sure that my session in the generator is rolled back here?
        # I tried passing cleanup_exit_stack_of_func(get_session, raise_exception=True) but it did not work.
    else:
        # this will finalize dependencies normally to trigger commit + close
        run_coroutine_sync(
            cleanup_exit_stack_of_func(get_some_service, raise_exception=False)
        )

How would i make sure that my session in the generator is rolled back here?

[JasperSui]

Hi @vincentleeuwen,

Yeah it's the right place to ask, would you try these two to see if one of them work?

Option A: treat some_service as a Depends parameter and use cleanup_all_exit_stacks with finally

from typing import Annotated
from fastapi_injectable import injectable, cleanup_all_exit_stacks, run_coroutine_sync

@celery_app.task  # type: ignore[misc]
@injectable
def run_demo_import_task(some_service: Annotated[SomeService, Depends(get_some_service)], file: io.BytesIO) -> None:
    try:
        some_service.some_task(file=file)
    except Exception:
        pass
    finally:
        run_coroutine_sync(cleanup_all_exit_stacks())

---

Option B: cleanup_exit_stack_of_func on get_some_service instead of get_session

@celery_app.task  # type: ignore[misc]
def run_demo_import_task(file: io.BytesIO) -> None:
    try:
        get_some_service().some_task(file=file)
    except Exception:
        run_coroutine_sync(cleanup_exit_stack_of_func(get_some_service, raise_exception=True))
    else:
        # this will finalize dependencies normally to trigger commit + close
        run_coroutine_sync(
            cleanup_exit_stack_of_func(get_some_service, raise_exception=False)
        )

---

I'm guessing the rollback() hasn't executed because of the wrong exit stack cleanup, if both don't work, could you provide reproducible steps?

BTW thanks for your kind word!

[vincentleeuwen]

@JasperSui thanks for the quick response! I was able to make ti work by storing sessoin in context and looking up that context later in the dependency tree (in Depends(get_session) basically). I also needed to disable caching as otherwise the next tesk woudl try to access the wrong session.