FastAPI TypeError: retreive_job() got multiple values for argument 'id'

Question

I'm trying to build a simple job board using Python, FastAPI and Async sqlalchemy by following the official FastAPI documentation. I'm having some issues regarding retrieving the data from database by ID.

The Following is hopefully a minimum reproducible code segment:

schemas/jobs.py

from typing import Optional
from pydantic import BaseModel
from datetime import date, datetime


class JobBase(BaseModel):
    title: Optional[str] = None
    company_name: Optional[str] = None
    company_url: Optional[str] = None
    location: Optional[str] = "remote"
    description: Optional[str] = None
    date_posted: Optional[date] = datetime.now().date()

class JobCreate(JobBase):
    title: str
    company_name: str
    location: str
    description: str

class ShowJob(JobBase):
    title: str
    company_name: str
    company_url: Optional[str]
    location: str
    date_posted: date
    description: str

    class Config():
        orm_mode = True

routes/route_jobs.py

from fastapi import APIRouter, HTTPException, status
from fastapi import Depends

from db.repository.job_board_dal import job_board
from schemas.jobs import JobCreate, ShowJob
from db.repository.job_board_dal import Job
from depends import get_db

router = APIRouter()

@router.post("/create-job",response_model=ShowJob)
async def create_user(Job: JobCreate, jobs: Job = Depends(get_db)):
    owner_id = 1
    return await jobs.create_new_job(Job, owner_id)

@router.get("/get/{id}")
def retrieve_job_by_id(id:int, job_board = Depends(get_db)):
    #print(type(session))
    job_id = job_board.retrieve_job(job_board, id=id)
    if not job_id:
        HTTPException(status_code=status.HTTP_404_NOT_FOUND,
        detail=f"Job with id {id} does not exist")
    return job_id

db/repository/job_board_dal.py

from sqlalchemy.orm import Session, query

from schemas.users import UserCreate
from schemas.jobs import JobCreate
from db.models.users import User
from db.models.jobs import Job
from core.hashing import Hasher



class job_board():
    def __init__(self, db_session: Session):
        self.db_session = db_session

    async def register_user(self, user: UserCreate):
        new_user = User(username=user.username,
        email=user.email,
        hashed_password=Hasher.get_password_hash(user.password),
        is_active = False,
        is_superuser=False
        )
        self.db_session.add(new_user)
        await self.db_session.flush()
        return new_user

    
    async def create_new_job(self, job: JobCreate, owner_id: int):
        new_job = Job(**job.dict(), owner_id = owner_id)
        self.db_session.add(new_job)
        await self.db_session.flush()
        return new_job

    def retrieve_job(self, id:int):
        item = self.db_session.query(Job).filter(Job.id == id).first()
        return item

depends.py

from db.session import async_session
from db.repository.job_board_dal import job_board

async def get_db():
    async with async_session() as session:
        async with session.begin():
            yield job_board(session)

I tried to change the dependency

@router.get("/get/{id}")
def retrieve_job_by_id(id:int, id_job: job_board = Depends(get_db)):
    #print(type(session))
    job_id = job_board.retrieve_job(id_job, id=id)
    if not job_id:
        HTTPException(status_code=status.HTTP_404_NOT_FOUND,
        detail=f"Job with id {id} does not exist")
    return job_id

it gives me this error AttributeError: 'AsyncSession' object has no attribute 'query'. Any help would be much appreciated.

Answer 1

AsyncSession class doesn't declare a query property.

You can use AsyncSession.get the fetch the Job record matching a given id. This method return None if no record is found matching the said id.

For example,

async def retrieve_job(self, id:int):
    item = await self.db_session.get(Job, id)
    return item