FastAPI Backend Development Bootcamp · บทเรียน

การเข้าถึงฐานข้อมูลแบบอะซิงโครนัส

สำรวจไดรเวอร์ฐานข้อมูลและ ORM แบบอะซิงโครนัส เช่น `asyncpg` และ `SQLModel` สำหรับการทำงานกับฐานข้อมูลโดยไม่บล็อก

การเข้าถึงฐานข้อมูลแบบอะซิงโครนัส เป็นบทเรียน FastAPI Backend Development Bootcamp ฟรีบน CoddyKit นี่คือบทเรียนที่ 2 จากทั้งหมด 4 บทเรียน คุณสามารถอ่านบทเรียนทั้งหมดด้านล่างฟรี — จากนั้นลองปฏิบัติด้วยตัวคุณเองในเบราว์เซอร์พร้อมตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7 บทเรียนนี้เป็นส่วนหนึ่งของเส้นทางการเรียน FastAPI Backend Development Bootcamp และความก้าวหน้าของคุณจะซิงค์ข้ามเว็บและแอป CoddyKit คอร์ส FastAPI Backend Development Bootcamp มีบทเรียนทั้งหมด 4 บทเรียน

บางส่วนของบทเรียนนี้ยังไม่ได้รับการแปล และแสดงเป็นภาษาอังกฤษ

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!

คำถามที่พบบ่อย

บทเรียน “การเข้าถึงฐานข้อมูลแบบอะซิงโครนัส” ฟรีหรือไม่

ใช่ — ข้อความเต็มของ “การเข้าถึงฐานข้อมูลแบบอะซิงโครนัส” ฟรีให้อ่านที่นี่บนเว็บ เพื่อปฏิบัติแบบโต้ตอบ (ตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7) และปลดล็อคส่วนที่เหลือของคอร์ส FastAPI Backend Development Bootcamp ให้อัปเกรดเป็น CoddyKit PRO คอร์ส FastAPI Backend Development Bootcamp มีบทเรียนทั้งหมด 4 บทเรียน

คุณจะเรียนรู้อะไรในบทเรียน “การเข้าถึงฐานข้อมูลแบบอะซิงโครนัส”

สำรวจไดรเวอร์ฐานข้อมูลและ ORM แบบอะซิงโครนัส เช่น `asyncpg` และ `SQLModel` สำหรับการทำงานกับฐานข้อมูลโดยไม่บล็อก คุณปฏิบัติ FastAPI Backend Development Bootcamp ด้วยโค้ดที่ใช้งานได้จริงที่คุณเรียกใช้โดยตรงในเบราว์เซอร์ และติวเตอร์ AI ตลอด 24/7 ตอบคำถามของคุณขณะที่คุณไปผ่านบทเรียน

คุณต้องมีประสบการณ์ก่อนที่จะเริ่มเรียน FastAPI Backend Development Bootcamp หรือไม่

ไม่จำเป็นต้องมีประสบการณ์มาก่อน FastAPI Backend Development Bootcamp บน CoddyKit ออกแบบมาสำหรับผู้เริ่มต้นไปจนถึงผู้เรียนขั้นสูง คุณสามารถเริ่มต้นที่นี่หรือเริ่มจากตัวแรกและเรียนด้วยความเร็วของคุณเอง นี่คือบทเรียนที่ 2 จากทั้งหมด 4 บทเรียน

บทเรียน “การเข้าถึงฐานข้อมูลแบบอะซิงโครนัส” ใช้เวลานานแค่ไหน

บทเรียน CoddyKit ส่วนใหญ่ใช้เวลาประมาณ 5–10 นาที แต่ละบทเรียนจึงสั้นและเป็นแบบโต้ตอบ คุณสามารถก้าวหน้าอย่างต่อเนื่องและกลับมาเรียนต่อจากตรงที่เพิ่งหยุดบนเว็บและแอปได้เลย

ฉันเขียนและรันโค้ดในบทเรียน FastAPI Backend Development Bootcamp นี้ได้ไหม

ได้ บทเรียน FastAPI Backend Development Bootcamp ทุกบทมีตัวแก้ไขโค้ดในตัว คุณจึงเขียนและรันโค้ดจริงได้เลยในเบราว์เซอร์ และได้รับข้อเสนอแนะจาก AI ในทันที — ไม่ต้องติดตั้งในเครื่องของคุณ

บทเรียนทั้งหมดในหลักสูตรนี้

← กลับไปที่ FastAPI Backend Development Bootcamp