从 SQLAlchemy declarative_base 定义的表中选择答案

【问题标题】：Select from table defined by SQLAlchemy declarative_base从 SQLAlchemy declarative_base 定义的表中选择
【发布时间】：2020-08-26 00:22:46
【问题描述】：

我正在学习 FastAPI 教程，我正在尝试使用 SQLAlchemy+Alembic+databases 创建表。
在我的main.py 我有：

from typing import List

import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel
from sqlalchemy import Table

DATABASE_URL = "sqlite:///./test.db"

database = databases.Database(DATABASE_URL)
metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    "note",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)

class Note2(BaseModel):
    id: int
    text: str
    completed: bool

app = FastAPI()

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

@app.get("/notes/", response_model=List[Note2])
async def read_notes():
    query = notes.select()
    return await database.fetch_all(query)

这行得通 - 我可以GET /notes/ 端点。但是在具有端点的同一模块中创建数据库表看起来很新手，所以我决定制作models.py 文件并在那里创建一个普通模型，如下所示：

import sqlalchemy as sa
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Note(Base):
    __tablename__ = "note"

    id = sa.Column(sa.Integer, primary_key=True)
    text = sa.Column(sa.String)
    completed = sa.Column(sa.Boolean)

这里出现了一个问题 - 当我像这样更改端点时：

from app_second.models import Note

@app.get("/notes/", response_model=List[Note2])
async def read_notes():
    query = Note().select()
    return await database.fetch_all(query)

我收到一个错误：

AttributeError: 'Note' 对象没有属性 'select'

正如上面提到的here - declarative_base() 只是Table + mapper 的语法 shugar。但是以这种方式声明的选择/过滤/更新表的正确方法是什么？

【问题讨论】：

Note 是一个 ORM 模型，所以你必须做session.query(Note).filter(...)。 note 是Table，可以通过Note.__table__ 访问
@snakecharmerb 是的，我认为它应该适用于正常的 SQLALchemy 流程，但我正在尝试使用 databases 使代码异步 - 它不使用 session（至少在教程中）像我一样制作）。
@snakecharmerb 但看起来Note.__table__.select() 也可以。 :) 但这是最好的方法吗？通常，在 python 代码中使用 dunder 方法/属性看起来不是一个好方法。
这个 tutorial 似乎涵盖了使用带有 FastApi 的声明式样式模型
@snakecharmerb 但正如我所见 - 端点是同步的，没有 async def ... 对我来说，这部分教程看起来就像是对同步数据库连接的解释......

标签： python sqlalchemy fastapi

【解决方案1】：

您可以使用 table 属性来访问表方法，然后使用 database.fetch_all() 或类似方法。例如：

from sqlalchemy import select
    
...
skip = 0
limit = 100

query = (
    Note.__table__.select()
    .offset(skip)
    .limit(limit)
)
return await database.fetch_all(query)

【讨论】：