Helden mit Limit und Offset mit FastAPI lesen

Wenn ein Client eine Anfrage sendet, um alle Helden zu erhalten, haben wir bisher alle zurückgegeben.

Aber wenn wir Tausende von Helden hätten, könnte das viele Rechenressourcen, Bandbreite usw. verbrauchen.

Wir wollen es also wahrscheinlich begrenzen.

Wir verwenden den gleichen Offset und Limit, den wir in den vorherigen Tutorial-Kapiteln für die API gelernt haben.

Info

In vielen Fällen wird dies auch als Paginierung bezeichnet.

Limit und Offset zu den Abfrageparametern hinzufügen

Fügen wir limit und offset zu den Abfrageparametern hinzu.

Standardmäßig geben wir die ersten Ergebnisse aus der Datenbank zurück, sodass offset einen Standardwert von 0 hat.

Und standardmäßig geben wir maximal 100 Helden zurück, sodass limit einen Standardwert von 100 hat.

Python 3.10+

from fastapi import FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select

# Code here omitted 👈

@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
    with Session(engine) as session:
        heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
        return heroes

# Code below omitted 👇

👀 Vorschau der vollständigen Datei

Python 3.10+

from fastapi import FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: int | None = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: int | None = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
    with Session(engine) as session:
        heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
        return heroes


@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int):
    with Session(engine) as session:
        hero = session.get(Hero, hero_id)
        if not hero:
            raise HTTPException(status_code=404, detail="Hero not found")
        return hero

🤓 Andere Versionen und Varianten

Python 3.9+Python 3.8+

from typing import Optional

from fastapi import FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
    with Session(engine) as session:
        heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
        return heroes


@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int):
    with Session(engine) as session:
        hero = session.get(Hero, hero_id)
        if not hero:
            raise HTTPException(status_code=404, detail="Hero not found")
        return hero

from typing import List, Optional

from fastapi import FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, select


class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)


class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)


class HeroCreate(HeroBase):
    pass


class HeroPublic(HeroBase):
    id: int


sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)


def create_db_and_tables():
    SQLModel.metadata.create_all(engine)


app = FastAPI()


@app.on_event("startup")
def on_startup():
    create_db_and_tables()


@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate):
    with Session(engine) as session:
        db_hero = Hero.model_validate(hero)
        session.add(db_hero)
        session.commit()
        session.refresh(db_hero)
        return db_hero


@app.get("/heroes/", response_model=List[HeroPublic])
def read_heroes(offset: int = 0, limit: int = Query(default=100, le=100)):
    with Session(engine) as session:
        heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()
        return heroes


@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int):
    with Session(engine) as session:
        hero = session.get(Hero, hero_id)
        if not hero:
            raise HTTPException(status_code=404, detail="Hero not found")
        return hero

Wir möchten es den Clients ermöglichen, unterschiedliche offset- und limit-Werte festzulegen.

Aber wir möchten nicht, dass sie ein limit von etwas wie 9999 festlegen können, das ist über 9000! 😱

Um dies zu verhindern, fügen wir eine zusätzliche Validierung für den limit-Abfrageparameter hinzu und deklarieren, dass er kleiner oder gleich 100 sein muss, mit le=100.

Auf diese Weise kann ein Client entscheiden, weniger Helden zu nehmen, wenn er möchte, aber nicht mehr.

Info

Wenn Sie sich auffrischen müssen, wie Abfrageparameter und ihre Validierung funktionieren, lesen Sie die Dokumentation in FastAPI.

• Abfrageparameter
• Abfrageparameter und Zeichenfolgenvalidierungen
• Pfadparameter und numerische Validierungen

Die Doku-Oberfläche überprüfen

Jetzt sehen wir, dass die Dokumentations-UI die neuen Parameter zur Steuerung von Limit und Offset unserer Daten anzeigt.

Zusammenfassung

Sie können die automatische Datenvalidierung von FastAPI verwenden, um die Parameter für limit und offset zu erhalten, und diese dann mit der Session verwenden, um Datenbereiche zu steuern, die in Antworten gesendet werden.