FastAPI AttributeError: 'job_board' object has no attribute 'query'

Question

我正在尝试使用 Python、FastAPI 和 Async sqlalchemy 构建一个简单的工作板，方法是遵循当我尝试通过 [=14= 检索工作时出现的官方 FastAPI documentation.Problem ] 从数据库中，当我到达 "/get/{id}" 端点时，它一直给我这个错误 AttributeError: 'job_board' object has no attribute 'query'。

以下是希望的最小可重现代码段：

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 sqlalchemy.orm.session import Session

from db.repository.job_board_dal import job_board
from db.models.jobs import Job as model_job
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 retreive_job_by_id(id:int, session: Session = Depends(get_db)):
    #print(type(session))
    job_id = job_board.retrieve_job(session, 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

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(db: Session, id:int):
        item = db.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)

我确实尝试了很多东西，我什至尝试用单独的异步会话检索，这也给了我 AttributeError：'AsyncSession' 对象没有属性 'query'。任何帮助将不胜感激。

Answer 1

get_db 不是 return 会话，它 return 是 job_board 的一个实例。因此，当您执行 retreive_job_by_id(id:int, session: Session = Depends(get_db)) 时，您会将会话设置为 class of job_board。因此，当您调用 job_id = job_board.retrieve_job(session, id=id) 时，您传递的会话是 job_board 的一个实例。因此，当您执行 item = db.query(Job) db 时，这里是 Job_board 的一个实例，而 job_board 没有名为查询的方法。相反，您可能只想更新检索作业以利用 job_board class 实例访问其会话。


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

在您的路由器中，您可以使其更具可读性，因为

@router.get("/get/{id}")
def retreive_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

FastAPI AttributeError: 'job_board' object has no attribute 'query'

FastAPI AttributeError: 'job_board' object has no attribute 'query'

python

asynchronous

sqlalchemy

python-3.x

fastapi