I’m working on a FastAPI application where I’m using Dependency Injector to manage my dependencies. I have a JobService that interacts with the database through SQLAlchemy. I’m encountering an error when trying to create a new job: ‘method’ object does not support the context manager protocol.

Here is the structure of my code:

Database Module (database.py)

from contextlib import contextmanager
from sqlalchemy import create_engine, orm
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session
import urllib

Base = declarative_base()

class MySQLDatabase:
    def __init__(self, db_host, db_port, db_username, db_password, db_schema) -> None:
        self.__db_conn_string = f"""mysql+pymysql://{db_username}:%s@{db_host}:{db_port}/{db_schema}"""
        self._engine = create_engine(self.__db_conn_string % urllib.parse.quote_plus(db_password), echo=True)
        self._session_factory = orm.scoped_session(
            orm.sessionmaker(
                autocommit=False,
                autoflush=False,
                bind=self._engine,
            ),
        )

    @contextmanager
    def session(self):
        session: Session = self._session_factory()
        try:
            yield session
        except Exception:
            session.rollback()
            raise
        finally:
            session.close()

Job Service (jobs_service.py)

from contextlib import AbstractContextManager
from sqlalchemy.orm import Session
from typing import Callable
from app.repository.mysql.jobs_dao import JobsDao
from app.dtos.jobs import CreateJobRequestDto
from fastapi import HTTPException, status
from datetime import datetime

class JobService:
    def __init__(self, job_dao: JobsDao, mysql_session_factory: Callable[..., AbstractContextManager[Session]]) -> None:
        self.job_dao = job_dao
        self.session_factory = mysql_session_factory

    def create_new_job(self, title: str, description: str, organization_id: int, expiry_at: datetime) -> CreateJobRequestDto:
        try:
            with self.session_factory() as session:
                new_job = self.job_dao.create_model(
                    session=session,
                    title=title,
                    description=description,
                    expiry_at=expiry_at,
                    organization_id=organization_id,
                )
                session.add(new_job)
                session.commit()
                session.refresh(new_job)

                return CreateJobRequestDto.from_orm(new_job)
        except Exception as e:
            raise HTTPException(
                status_code=status.HTTP_501_NOT_IMPLEMENTED, detail="Job not created"
            )

Dependency Injection Setup (containers.py)

from dependency_injector import containers, providers
from app.services.jobs_service import JobService
from app.repository.mysql.jobs_dao import JobsDao
from app.database import MySQLDatabase

class Container(containers.DeclarativeContainer):
    config = providers.Configuration()

    mysql_db = providers.Singleton(
        MySQLDatabase,
        db_host=config.properties.database.mysql.host,
        db_port=config.properties.database.mysql.port,
        db_username=config.properties.database.mysql.user,
        db_password=config.properties.database.mysql.password,
        db_schema=config.properties.database.mysql.schema,
    )

    job_dao = providers.Factory(JobsDao)
    job_service = providers.Factory(
        JobService,
        job_dao=job_dao,
        mysql_session_factory=mysql_db.provided.session,
    )

Router (router.py)

from fastapi import APIRouter, Depends, HTTPException, status
from app.controller.containers import Container
from app.dtos.jobs import CreateJobResponseDto, CreateJobRequestDto
from dependency_injector.wiring import Provide, inject
from fastapi.security import OAuth2PasswordBearer
from app.controller.dependencies import validate_user
from app.services.jobs_service import JobService

router = APIRouter(prefix="/jobs")

@router.post("/", summary="Create a new job", response_model=CreateJobResponseDto)
@inject
async def create_new_job(
    job: CreateJobRequestDto,
    authenticated_user: OAuth2PasswordBearer = Depends(validate_user),
    job_services: JobService = Depends(Provide[Container.job_service]),
):
    if not authenticated_user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token expired",
            headers={"WWW-Authenticate": "Bearer"},
        )

    if not hasattr(authenticated_user, "organization_id") or authenticated_user.organization_id is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Login with organization account to continue",
            headers={"WWW-Authenticate": "Bearer"},
        )

    return job_services.create_new_job(
        title=job.title,
        description=job.description,
        expiry_at=job.expiry_at,
        organization_id=authenticated_user.organization_id
    )

When I try to create a new job, I get the following error:

'method' object does not support the context manager protocol
  File "path_to_projectapproutersjobs.py", line 48, in create_new_job
    return job_services.create_new_job(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "path_to_projectappservicesjobs_service.py", line 21, in create_new_job
    with self.session_factory() as session:
TypeError: 'method' object does not support the context manager protocol

I am not sure what is causing this issue. Any guidance on how to resolve this would be greatly appreciated.