Python Academy · leksjon

Asynkrone endepunkter og databaseintegrasjon

Skriv asynkrone rutehåndterere og koble til en database med SQLAlchemy.

Leksjon 4 av 413 trinn

Asynkrone endepunkter og databaseintegrasjon er en gratis leksjon i Python Academy på CoddyKit. Dette er leksjon 4 av 4. Du kan lese hele leksjonen gratis nedenfor – og deretter øve praktisk i nettleseren med en innebygd kodeeditor og en AI-veileder som er tilgjengelig døgnet rundt. Den er en del av læringsløpet i Python Academy, og fremdriften din synkroniseres mellom nettet og CoddyKit-appen. Kurset i Python Academy inneholder totalt 4 leksjoner.

Endepunkter med async def

Deklarer rutebehandlere med async def for ikke-blokkerende I/O. Bruk def for CPU-intensive eller blokkerende operasjoner (FastAPI kjører dem i en trådpool).

from fastapi import FastAPI
import asyncio
app = FastAPI()

@app.get("/slow")
async def slow():
    await asyncio.sleep(1)   # non-blocking
    return {"done": True}

asyncpg for PostgreSQL

asyncpg er en rask, asynkron PostgreSQL-driver. Bruk en tilkoblingspool som deles mellom forespørsler.

# pip install asyncpg
import asyncpg, asyncio

async def main():
    pool = await asyncpg.create_pool("postgresql://user:pass@host/db")
    async with pool.acquire() as conn:
        rows = await conn.fetch("SELECT id, name FROM users")
    await pool.close()
    return rows

Asynkron SQLAlchemy

SQLAlchemy 1.4+ med async_sessionmaker tilbyr en asynkron ORM. Bruk AsyncSession som en FastAPI-avhengighet.

# pip install sqlalchemy aiosqlite
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

engine = create_async_engine("sqlite+aiosqlite:///db.sqlite")
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

Databaseavhengighet

Returner sesjonen fra en FastAPI-avhengighet, slik at den bekreftes og lukkes etter forespørselen.

from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession
app = FastAPI()

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    return result.scalars().all()

SQLAlchemy ORM-modeller

Definer ORM-modeller med DeclarativeBase og knytt dem til databasetabeller.

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"

    id:    Mapped[int] = mapped_column(primary_key=True)
    name:  Mapped[str]
    email: Mapped[str]

CRUD-operasjoner

Implementer grunnleggende Create, Read, Update, Delete ved hjelp av asynkrone sesjonsmetoder i SQLAlchemy.

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

async def create_user(db: AsyncSession, name: str, email: str):
    user = User(name=name, email=email)
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

async def get_user(db: AsyncSession, user_id: int):
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

Alembic for migreringer

Bruk Alembic til å håndtere migreringer av databaseskjemaet på en versjonskontrollert måte.

# pip install alembic
# alembic init alembic
# Edit alembic/env.py to point to your models

# Generate a migration:
# alembic revision --autogenerate -m "add users table"

# Apply:
# alembic upgrade head

httpx for asynkrone HTTP-kall

Bruk httpx.AsyncClient til å kalle eksterne API-er fra asynkrone FastAPI-endepunkter.

# pip install httpx
import httpx
from fastapi import FastAPI
app = FastAPI()

@app.get("/weather")
async def weather(city: str):
    async with httpx.AsyncClient() as client:
        r = await client.get(f"https://wttr.in/{city}?format=j1")
        return r.json()

Bakgrunnsoppgaver

Bruk BackgroundTasks til å utføre arbeid etter at svaret er sendt – for eksempel sende en e-post eller oppdatere en hurtigbuffer.

from fastapi import FastAPI, BackgroundTasks
app = FastAPI()

def send_welcome_email(email: str):
    # blocking I/O — runs in background thread
    smtp_send(email, "Welcome!")

@app.post("/register")
def register(email: str, bg: BackgroundTasks):
    create_user(email)
    bg.add_task(send_welcome_email, email)
    return {"status": "registered"}

Lifespan-hendelser

Bruk lifespan-kontekstbehandleren (FastAPI 0.93+) til å kjøre kode ved oppstart og avslutning – for eksempel opprette databasepoolen.

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.pool = await asyncpg.create_pool(DSN)
    yield
    await app.state.pool.close()

app = FastAPI(lifespan=lifespan)

Testing av FastAPI med httpx

Bruk httpx.AsyncClient sammen med ASGITransport til å teste FastAPI-endepunkter uten å kjøre en server.

import pytest, httpx
from fastapi import FastAPI
app = FastAPI()

@app.get("/ping")
async def ping(): return {"pong": True}

@pytest.mark.asyncio
async def test_ping():
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app),
        base_url="http://test"
    ) as client:
        r = await client.get("/ping")
    assert r.json() == {"pong": True}

Hurtigsjekk

Når bør man bruke async def i stedet for def for en rutebehandler i FastAPI?

Besvar den i den interaktive leksjonen

Oppsummering

Bruk endepunkter med async def sammen med asynkrone databasedrivere (asyncpg, asynkron SQLAlchemy). Håndter databasesesjonen som en FastAPI-avhengighet. Bruk BackgroundTasks til arbeid etter at svaret er sendt, og lifespan-kontekstbehandleren ved oppstart og avslutning.

Gratis å komme i gang

Lær deg Python med en AI-veileder – gratis

Skriv og kjør ekte kode i nettleseren, få umiddelbar hjelp fra en AI-veileder som er tilgjengelig døgnet rundt, og fortsett der du slapp – på nettet eller i appen.

Kurs: 76
Leksjoner: 320

Start læringen gratis Se hele kurset

Ofte stilte spørsmål

Er leksjonen «Asynkrone endepunkter og databaseintegrasjon» gratis?

Ja – hele teksten i «Asynkrone endepunkter og databaseintegrasjon» er gratis å lese her på nettet. For å øve interaktivt med en innebygd kodeeditor og en AI-veileder som er tilgjengelig døgnet rundt, og for å låse opp resten av Python Academy-kurset, kan du oppgradere til CoddyKit PRO. Kurset i Python Academy inneholder totalt 4 leksjoner.

Hva lærer jeg i «Asynkrone endepunkter og databaseintegrasjon»?

Skriv asynkrone rutehåndterere og koble til en database med SQLAlchemy. Du øver på Python Academy med praktisk kode som du kjører direkte i nettleseren, mens en AI-veileder som er tilgjengelig døgnet rundt, svarer på spørsmålene dine mens du jobber deg gjennom leksjonen.

Trenger jeg erfaring for å begynne med Python Academy?

Ingen tidligere erfaring er nødvendig. Python Academy på CoddyKit er lagt opp for både nybegynnere og viderekomne, så De kan begynne her eller helt fra start og lære i Deres eget tempo. Dette er leksjon 4 av 4.

Hvor lang tid tar leksjonen «Asynkrone endepunkter og databaseintegrasjon»?

De fleste CoddyKit-leksjoner tar omtrent 5–10 minutter. Hver leksjon er kort og interaktiv, slik at De gjør jevne fremskritt og kan fortsette akkurat der De slapp – både på nettet og i appen.

Kan jeg skrive og kjøre kode i denne Python Academy-leksjonen?

Ja. Alle Python Academy-leksjoner har en innebygd kodeeditor, slik at De kan skrive og kjøre ekte kode direkte i nettleseren og få umiddelbar tilbakemelding fra AI – uten lokal konfigurering.

Alle leksjonene i dette kurset

FastAPI-prosjekt og første endepunkt
Stiparametere, spørringsparametere og forespørselskropper
Avhengighetsinjeksjon og autentisering
Asynkrone endepunkter og databaseintegrasjon

← Tilbake til Python Academy