私は高速APIが初めてです。テーブルからユーザーを取得しようとしていますが、次のエラーが発生します。
user = await db.query(User).filter(User.login == data['login']).first()
AttributeError: 'Depends' オブジェクトに属性 'query' がありません
ここに私のserver/database.pyがあります
import os
from sqlmodel import SQLModel, Session
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
DATABASE_URL = os.environ.get("DATABASE_URL")
engine = create_async_engine(DATABASE_URL, echo = True, future = True)
async def init_db():
async with engine.begin() as conn:
# await conn.run_sync(SQLModel.metadata.drop_all)
await conn.run_sync(SQLModel.metadata.create_all) # init is ok, db table working
pass
async def get_session() -> AsyncSession:
async_session = sessionmaker(
engine, class_ = AsyncSession,
expire_on_commit = False
)
async with async_session() as session:
yield session
ここにserver/web_server.pyがあります
from fastapi import Depends, FastAPI
from sqlalchemy.future import select
from fastapi_socketio import SocketManager
from fastapi_login import LoginManager
import json
from database import get_session, init_db
from models import User, File
app = FastAPI()
sio = SocketManager(app = app)
@app.on_event("startup")
async def on_startup():
await init_db() # is ok
@app.sio.on('connect')
async def handle_connect(sid, connection_data):
await app.sio.emit('connect', 'User joined') # is ok
@app.sio.on('authorize')
async def authorize(sid, message, db = Depends(get_session)):
data = json.loads(message) # is ok
user = await db.query(User).filter(User.login == data['login']).first() # error in this line
print(user)
if __name__ == '__main__':
import uvicorn
uvicorn.run("web_server:app", host = '0.0.0.0', port = 8000, reload = True, debug = False)
アプリに socketio を使用しています。socketio.on('authorize')メッセージにはdata = {'login' : '...', 'password' : '...'}が含まれています。パスワードを確認して承認するために、私のpostgresqlテーブルのユーザー。したがって、「get_session」、クエリ、実行、コミットに問題があり、同じエラーAttributeError: 'Depends' object has no attribute .. が発生します。