Asynkroniset päätepisteet ja tietokantaintegraatio
Kirjoittakaa asynkronisia reittikäsittelijöitä ja yhdistäkää sovellus tietokantaan SQLAlchemyn avulla.
Asynkroniset päätepisteet ja tietokantaintegraatio on ilmainen Python Academy-oppitunti CoddyKitissä. Tämä on oppitunti 4/4. Voit lukea koko oppitunnin alta ilmaiseksi ja harjoitella sen jälkeen käytännössä selaimessa sisäänrakennetulla koodieditorilla ja ympäri vuorokauden käytettävissä olevan tekoälytuutorin avulla. Oppitunti kuuluu Python Academy-oppimispolkuun, ja edistymisesi synkronoituu verkon ja CoddyKit-sovelluksen välillä. Python Academy-kurssilla on yhteensä 4 oppituntia.
async def -päätepisteet
Määritelkää reitinkäsittelijät async def-syntaksilla ei-blokkaavaa I/O:ta varten. Käyttäkää def-syntaksia suorittimeen sidotuissa tai blokkaavissa toiminnoissa (FastAPI suorittaa ne säiepoolissa).
from fastapi import FastAPI
import asyncio
app = FastAPI()
@app.get("/slow")
async def slow():
await asyncio.sleep(1) # non-blocking
return {"done": True}asyncpg PostgreSQL:ää varten
asyncpg on nopea, asynkroninen PostgreSQL-ajuri. Käyttäkää pyyntöjen kesken jaettua yhteyspoolia.
# 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 rowsSQLAlchemy Async
SQLAlchemy 1.4+ ja async_sessionmaker tarjoavat asynkronisen ORM:n. Käyttäkää AsyncSession-istuntoa FastAPI-riippuvuutena.
# 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 sessionTietokantariippuvuus
Palauttakaa istunto FastAPI-riippuvuudesta yieldillä, jotta se commitoidaan ja suljetaan pyynnön jälkeen.
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 -mallit
Määritelkää ORM-mallit DeclarativeBase-luokan avulla ja yhdistäkää ne tietokantatauluihin.
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-toiminnot
Toteuttakaa perustason luonti-, luku-, päivitys- ja poistotoiminnot SQLAlchemyn asynkronisten istuntometodien avulla.
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 migraatioita varten
Käyttäkää Alembicia tietokannan skeeman migraatioiden hallintaan versionhallinnan avulla.
# 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 headhttpx asynkronisia HTTP-kutsuja varten
Käyttäkää httpx.AsyncClient-asiakasohjelmaa ulkoisten API-rajapintojen kutsumiseen asynkronisista FastAPI-päätepisteistä.
# 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()Taustatehtävät
Käyttäkää BackgroundTasks-komponenttia työn suorittamiseen vastauksen lähettämisen jälkeen — esimerkiksi sähköpostin lähettämiseen tai välimuistin päivittämiseen.
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-tapahtumat
Käyttäkää lifespan-kontekstinhallintaa (FastAPI 0.93+) käynnistys- ja sammutuskoodin suorittamiseen — esimerkiksi tietokannan yhteyspoolin luomiseen.
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)FastAPI-sovellusten testaaminen httpx:llä
Käyttäkää httpx.AsyncClient-asiakasohjelmaa yhdessä ASGITransport-komponentin kanssa FastAPI-päätepisteiden testaamiseen ilman palvelimen käynnistämistä.
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}Pikatarkistus
Milloin FastAPI-reitinkäsittelijässä kannattaa käyttää async def-syntaksia ja milloin def-syntaksia?
Kertaus
Käyttäkää async def-päätepisteitä yhdessä asynkronisten tietokanta-ajureiden (asyncpg, SQLAlchemy async) kanssa. Hallitkaa tietokantaistuntoa FastAPI-riippuvuutena. Käyttäkää BackgroundTasks-komponenttia vastauksen jälkeiseen työhön ja lifespan-kontekstinhallintaa käynnistykseen ja sammutukseen.
Opi Python tekoälytuutorin avulla — ilmaiseksi
Kirjoita ja suorita oikeaa koodia selaimessa, saa välitöntä apua tekoälytuutorilta ympäri vuorokauden ja jatka siitä, mihin jäit, verkossa tai sovelluksessa.
- Kurssit
- 76
- Oppitunnit
- 320
Usein kysytyt kysymykset
Onko oppitunti ”Asynkroniset päätepisteet ja tietokantaintegraatio” ilmainen?
Kyllä – oppitunnin ”Asynkroniset päätepisteet ja tietokantaintegraatio” koko tekstin voi lukea täällä verkossa ilmaiseksi. Jos haluat harjoitella interaktiivisesti sisäänrakennetulla koodieditorilla ja ympäri vuorokauden käytettävissä olevan tekoälytuutorin avulla sekä avata koko Python Academy-kurssin, päivitä CoddyKit PROhon. Python Academy-kurssilla on yhteensä 4 oppituntia.
Mitä opin oppitunnilla ”Asynkroniset päätepisteet ja tietokantaintegraatio”?
Kirjoittakaa asynkronisia reittikäsittelijöitä ja yhdistäkää sovellus tietokantaan SQLAlchemyn avulla. Harjoittelet Python Academy-aihetta koodilla, jonka suoritat suoraan selaimessa. Ympäri vuorokauden käytettävissä oleva tekoälytuutori vastaa kysymyksiisi oppitunnin aikana.
Tarvitsenko kokemusta aloittaakseni Python Academy-opiskelun?
Aiempi kokemus ei ole tarpeen. CoddyKitin Python Academy-oppimispolku sopii vasta-alkajista edistyneisiin, joten voit aloittaa tästä tai alusta ja edetä omaan tahtiisi. Tämä on oppitunti 4/4.
Kuinka kauan ”Asynkroniset päätepisteet ja tietokantaintegraatio”-oppitunnin suorittaminen kestää?
Useimmat CoddyKitin oppitunnit kestävät noin 5–10 minuuttia. Jokainen oppitunti on lyhyt ja interaktiivinen, joten edistyt tasaisesti ja voit jatkaa siitä, mihin jäit – sekä verkossa että sovelluksessa.
Voinko kirjoittaa ja suorittaa koodia tällä Python Academy-oppitunnilla?
Kyllä. Jokainen Python Academy-oppitunti sisältää sisäänrakennetun koodieditorin, joten voit kirjoittaa ja suorittaa oikeaa koodia suoraan selaimessa ja saada välitöntä palautetta tekoälyltä – paikallista asennusta ei tarvita.
Kaikki tämän kurssin oppitunnit
- FastAPI-projektin määritys ja ensimmäinen päätepiste
- Polkuparametrit, kyselyparametrit ja pyyntörungot
- Riippuvuuksien injektointi ja todentaminen
- Asynkroniset päätepisteet ja tietokantaintegraatio