FastAPI AttributeError：“job_board”对象没有属性“查询”

Question

我正在尝试按照官方 FastAPI 文档使用 Python、FastAPI 和 Async sqlalchemy 构建一个简单的工作板。当我尝试通过 ID 从数据库中检索工作时出现问题，它一直给我这个当我点击"/get/{id}"endpoint 时出现错误AttributeError: 'job_board' object has no attribute 'query'。

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

架构/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

取决于.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' object has no attribute 'query'。 任何帮助将不胜感激。

Answer 1

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

    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