FastAPI Backend Development Bootcamp · Aula

Acesso assíncrono ao banco de dados

Explore controladores e ORMs assíncronos, como `asyncpg` e `SQLModel`, para realizar operações de banco de dados sem bloqueio.

Acesso assíncrono ao banco de dados é uma aula grátis de FastAPI Backend Development Bootcamp no CoddyKit. Esta é a aula 2 de 4. Você pode ler a aula completa abaixo gratuitamente — depois pratica ao vivo no navegador com um editor de código integrado e um tutor de IA 24/7. Faz parte do caminho de aprendizado de FastAPI Backend Development Bootcamp, e seu progresso é sincronizado entre a web e o app CoddyKit. O curso de FastAPI Backend Development Bootcamp inclui 4 aulas no total.

Partes desta aula ainda não foram traduzidas e aparecem em inglês.

Why Async DB Access?

When building high-performance web applications, especially with frameworks like FastAPI, database operations can often become a bottleneck. Traditional database calls are "blocking", meaning your application waits for the database to respond before moving on.

Asynchronous database access allows your application to perform other tasks while waiting for the database, preventing your API from becoming unresponsive under heavy load. This is crucial for scalability!

Sync vs. Async DB Calls

Imagine a restaurant where one chef handles everything. If a customer orders a complex dish (a database query), the chef stops all other work until that dish is complete. This is synchronous I/O.

In an asynchronous setup, the chef starts the complex dish, then immediately moves to prepare simpler dishes or take new orders while the complex dish cooks in the background. When the complex dish is ready, the chef picks it up. This non-blocking approach boosts efficiency!

Meet asyncpg: The Async Driver

For PostgreSQL, the go-to asynchronous driver in Python is asyncpg. It's a high-performance library specifically designed to work with Python's asyncio framework.

Fast: Written partly in C for speed.
Asynchronous: Fully non-blocking.
Direct: Provides a direct interface to PostgreSQL.

It's often used as the underlying driver for async ORMs or when you need fine-grained control.

asyncpg in Action

Let's see a basic example of connecting to a PostgreSQL database and running a simple query using asyncpg. Remember to replace placeholder credentials with your own!

import asyncio
import asyncpg

async def main():
    conn = None
    try:
        conn = await asyncpg.connect(user='postgres', password='mysecretpassword',
                                     database='testdb', host='localhost')
        print("Connected to PostgreSQL!")
        
        # Execute a query
        result = await conn.fetchval('SELECT 1 + 1')
        print(f"Query result: {result}") # Should be 2
        
    except Exception as e:
        print(f"Error: {e}")
    finally:
        if conn:
            await conn.close()
            print("Connection closed.")

if __name__ == "__main__":
    asyncio.run(main())

Awaiting Database Calls

In the previous example, you saw the await keyword before asyncpg.connect() and conn.fetchval(). This is crucial for asynchronous operations.

await tells Python: "This operation might take time, so pause here and let other tasks run in the meantime."
When the database operation completes, the task resumes from where it left off.
This non-blocking wait is what makes your FastAPI application scalable.

SQLModel: Async ORM Power

While asyncpg gives you low-level control, an Object Relational Mapper (ORM) simplifies database interactions by mapping database tables to Python objects. SQLModel is a modern, async-first ORM built on:

Pydantic: For data validation and serialization.
SQLAlchemy: The powerful and mature Python SQL toolkit.

It lets you define models once and use them for both your API request/response and database schema!

Setting up SQLModel for Async

To use SQLModel asynchronously, you need an asynchronous database engine. This typically involves using an async driver like asyncpg (which SQLAlchemy can use via asyncio). Here's how you'd set up the engine:

from sqlmodel import create_engine, SQLModel
import asyncio

# Replace with your actual async PostgreSQL connection string
# The 'postgresql+asyncpg' part tells SQLAlchemy to use asyncpg
DATABASE_URL = "postgresql+asyncpg://postgres:mysecretpassword@localhost/testdb"

async def main():
    engine = create_engine(DATABASE_URL, echo=True)
    print("Async SQLModel engine created.")
    
    # In a real app, you'd usually create tables here
    # async with engine.begin() as conn:
    #     await conn.run_sync(SQLModel.metadata.create_all)
    
    # Just demonstrating engine creation for this example
    await engine.dispose()
    print("Engine disposed.")

if __name__ == "__main__":
    asyncio.run(main())

Your First SQLModel

Defining a model in SQLModel is super intuitive. You inherit from SQLModel and use Pydantic-like field declarations. This single definition creates both your database table schema and your API's data validation schema!

from typing import Optional
from sqlmodel import Field, SQLModel

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

# This model can now be used with FastAPI for request bodies
# and with SQLAlchemy for database interactions.
print("Hero model defined successfully!")

Async CRUD with SQLModel

Now let's perform a simple Create and Read operation using our Hero model and the async engine. We'll use AsyncSession from SQLAlchemy's ORM for database interactions.

from typing import Optional
from sqlmodel import Field, SQLModel, create_engine, Session, select
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
import asyncio

DATABASE_URL = "postgresql+asyncpg://postgres:mysecretpassword@localhost/testdb"
async_engine = create_async_engine(DATABASE_URL, echo=False)
AsyncSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=async_engine, class_=AsyncSession)

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

async def create_db_and_tables():
    async with async_engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
    print("Database tables created/updated.")

async def create_hero(hero: Hero):
    async with AsyncSessionLocal() as session:
        session.add(hero)
        await session.commit()
        await session.refresh(hero)
        print(f"Created hero: {hero.name} (ID: {hero.id})")
        return hero

async def get_heroes():
    async with AsyncSessionLocal() as session:
        statement = select(Hero)
        results = await session.execute(statement)
        heroes = results.scalars().all()
        print("\nAll Heroes:")
        for hero in heroes:
            print(f"- {hero.name} (Age: {hero.age})")
        return heroes

async def main():
    await create_db_and_tables()

    hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson", age=28)
    hero_2 = Hero(name="Spider-Boy", secret_name="Pedro Parqueador")
    hero_3 = Hero(name="Rusty-Man", secret_name="Tommy Sharp", age=48)

    await create_hero(hero_1)
    await create_hero(hero_2)
    await create_hero(hero_3)

    await get_heroes()
    await async_engine.dispose()

if __name__ == "__main__":
    asyncio.run(main())

Async DB Check

You've learned about the importance of asynchronous database access and explored tools like asyncpg and SQLModel. Let's test your understanding!

Recap: Async DB for Scale

Great job! You've successfully explored asynchronous database access.

We understood why non-blocking I/O is vital for high-performance FastAPI apps.
We introduced asyncpg as a low-level async PostgreSQL driver.
We learned about SQLModel, an async-first ORM combining Pydantic and SQLAlchemy.
We saw practical examples of setting up and performing CRUD operations with these tools.

Mastering async database interactions is a key step towards building truly scalable and responsive backend services!

Perguntas Frequentes

A aula “Acesso assíncrono ao banco de dados” é grátis?

Sim — o texto completo de “Acesso assíncrono ao banco de dados” é grátis para ler aqui na web. Para praticá-la interativamente (um editor de código integrado e um tutor de IA 24/7) e desbloquear o restante do curso de FastAPI Backend Development Bootcamp, atualize para CoddyKit PRO. O curso de FastAPI Backend Development Bootcamp inclui 4 aulas no total.

O que vou aprender em “Acesso assíncrono ao banco de dados”?

Explore controladores e ORMs assíncronos, como `asyncpg` e `SQLModel`, para realizar operações de banco de dados sem bloqueio. Você pratica FastAPI Backend Development Bootcamp com código prático que executa diretamente no navegador, e um tutor de IA 24/7 responde suas dúvidas enquanto trabalha na aula.

Preciso ter experiência prévia para começar FastAPI Backend Development Bootcamp?

Nenhuma experiência prévia é necessária. FastAPI Backend Development Bootcamp no CoddyKit é estruturado para alunos iniciantes até avançados, então você pode começar aqui ou desde o início e aprender no seu ritmo. Esta é a aula 2 de 4.

Quanto tempo leva a aula “Acesso assíncrono ao banco de dados”?

A maioria das aulas CoddyKit leva cerca de 5–10 minutos. Cada uma é compacta e interativa, então você faz progresso constante e retoma exatamente de onde parou entre web e app.

Posso escrever e executar código nesta aula de FastAPI Backend Development Bootcamp?

Sim. Cada aula de FastAPI Backend Development Bootcamp inclui um editor de código integrado, então você escreve e executa código real direto no navegador e recebe feedback de IA instantaneamente — nenhuma configuração local necessária.

Todas as aulas deste curso

Estratégias de cache com Redis
Acesso assíncrono ao banco de dados
Balanceamento de carga e monitoramento
Tarefas em segundo plano e filas de trabalhos

← Voltar para FastAPI Backend Development Bootcamp