0Pricing
Python Academy · Lezione

Endpoint asincroni e integrazione con il database

Scriva gestori di route asincroni e si connetta a un database con SQLAlchemy.

Endpoint asincroni e integrazione con il database è una lezione Python Academy gratuita su CoddyKit. Questa è la lezione 4 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento Python Academy, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso Python Academy include 4 lezioni in totale.

Endpoint con async def

Dichiari i gestori delle route con async def per l'I/O non bloccante. Utilizzi def per le operazioni vincolate dalla CPU o bloccanti (FastAPI le esegue in un pool di thread).

from fastapi import FastAPI
import asyncio
app = FastAPI()

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

asyncpg per PostgreSQL

asyncpg è un driver PostgreSQL asincrono e veloce. Utilizzi un pool di connessioni condiviso tra le richieste.

# 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

SQLAlchemy asincrono

SQLAlchemy 1.4+ con async_sessionmaker fornisce un ORM asincrono. Utilizzi AsyncSession come dipendenza FastAPI.

# 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

Dipendenza del database

Utilizzi yield per fornire la sessione da una dipendenza FastAPI, in modo che venga eseguito il commit e che la sessione venga chiusa al termine della richiesta.

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()

Modelli ORM di SQLAlchemy

Definisca i modelli ORM con DeclarativeBase e li associ alle tabelle del database.

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]

Operazioni CRUD

Implementi le operazioni di base di creazione, lettura, aggiornamento ed eliminazione utilizzando i metodi della sessione asincrona di 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 per le migrazioni

Utilizzi Alembic per gestire le migrazioni dello schema del database in modo controllato tramite il versionamento.

# 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 per le chiamate HTTP asincrone

Utilizzi httpx.AsyncClient per chiamare API esterne dagli endpoint asincroni di FastAPI.

# 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()

Attività in background

Utilizzi BackgroundTasks per eseguire attività dopo l'invio della risposta — ad esempio, inviare un'e-mail o aggiornare una cache.

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"}

Eventi del ciclo di vita

Utilizzi il context manager lifespan (FastAPI 0.93+) per eseguire il codice di avvio e arresto — ad esempio, creare il pool del database.

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)

Testare FastAPI con httpx

Utilizzi httpx.AsyncClient con ASGITransport per testare gli endpoint FastAPI senza avviare un 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}

Verifica rapida

Quando dovrebbe utilizzare async def invece di def per un gestore di route FastAPI?

Riepilogo

Utilizzi endpoint async def con driver di database asincroni (asyncpg, SQLAlchemy asincrono). Gestisca la sessione del database come dipendenza FastAPI. Utilizzi BackgroundTasks per le attività successive alla risposta e il context manager lifespan per il codice di avvio e arresto.

Domande Frequenti

La lezione «Endpoint asincroni e integrazione con il database» è gratuita?

Sì — il testo completo di «Endpoint asincroni e integrazione con il database» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso Python Academy, passa a CoddyKit PRO. Il corso Python Academy include 4 lezioni in totale.

Cosa imparerò in «Endpoint asincroni e integrazione con il database»?

Scriva gestori di route asincroni e si connetta a un database con SQLAlchemy. Eserciti Python Academy con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.

Ho bisogno di esperienza per iniziare Python Academy?

Non è richiesta alcuna esperienza precedente. Python Academy su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 4 di 4.

Quanto tempo richiede la lezione «Endpoint asincroni e integrazione con il database»?

La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.

Posso scrivere ed eseguire codice in questa lezione Python Academy?

Sì. Ogni lezione Python Academy include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.

Tutte le lezioni di questo corso

  1. Configurazione di un progetto FastAPI e primo endpoint
  2. Parametri del percorso, parametri query e body delle richieste
  3. Dependency injection e autenticazione
  4. Endpoint asincroni e integrazione con il database
← Torna a Python Academy