Perform an SQL 'database.exec()' search in a function not under a FastAPI decorator

I am trying to streamline a FastAPI endpoint being developed by my team, in which we are repeatedly registering items to an SQL database if they don't already exist in it.

The session was defined in one file as follows:

from functools import partial

from sqlmodel import Session, create_engine
from fastapi import Depends

def get_db_session() -> Session:  # type: ignore
    _url = "postgresql+psycopg2://some-path:8000"
    engine = create_engine(_url)
    with Session(engine) as session:
        try:
            yield session
        finally:
            session.close()

our_db_session = partial(get_db_session, some_config_file)
our_db: Session = Depends(our_db_session)

It is then imported for use in the file where the database lookups are happening:

from fastapi import APIRouter

from other_file import our_db

router = APIRouter()

@router.post("/register_file/")
def register_file(
    file_name: Path,
    search_criteria: Union[int, str],
    db: Session = our_db,
):
    db_entry = db.exec(
        select(Table1)
        .where(Table1.file_name==file_name)
        .where(Table1.some_value==search_criteria)
    ).one()

When placed inside the decorated function, the lookup works without issue. However, as there are many such functions registering different things, and the lookup criteria is the same, it got repetitive typing the same thing over and over again within the decorated functions. As such, I tried to move the lookup function into a separate function that gets called within the decorated functions:

def get_db_entry(
    file_name: Path,
    search_criteria: Union[str, int],
    db: Session = our_db,
):
    db_entry = db.exec(
        select(Table1)
        .where(Table1.file_name==file_name)
        .where(Table1.some_value==search_criteria)
    ).one()
    return db_entry

@router.post("/register_file/")
def register_file(
    file_name: Path,
    search_criteria: Union[int, str],
    db: Session = our_db,
):
    db_entry = get_db_entry(file_name=file_name, search_criteria=search_criteria)

However, when I do it like that, I get an error saying that "Depends" does not have the attribute ".exec()".

I'm very new to FastAPI and SQL, so I would greatly appreciate advice on why this doesn't work, and what I need to change in order to migrate the search function out of the decorated function.

You should remove = our_db from the signature of get_db_entry function and pass this parameter explicitly when you call it:

def get_db_entry(
    file_name: Path,
    search_criteria: Union[str, int],
    db: Session,
):
    db_entry = db.exec(
        select(Table1)
        .where(Table1.file_name==file_name)
        .where(Table1.some_value==search_criteria)
    ).one()
    return db_entry

@router.post("/register_file/")
def register_file(
    file_name: Path,
    search_criteria: Union[int, str],
    db: Session = our_db,
):
    db_entry = get_db_entry(file_name=file_name, search_criteria=search_criteria, db=db)