FastAPI ناهمگام، SQLAlchemy ناهمگام و lifespan

۱. FastAPI هر endpoint را به یکی از دو روش اجرا می‌کند

app = FastAPI()


@app.get('/async-good')
async def async_good() -> dict[str, str]:
    await asyncio.sleep(1)  # pretend: await an async DB query or HTTP call
    return {'endpoint': 'async-good'}


@app.get('/sync-ok')
def sync_ok() -> dict[str, str]:
    time.sleep(1)  # pretend: a blocking library call -- fine in a thread
    return {'endpoint': 'sync-ok'}


@app.get('/async-bad')
async def async_bad() -> dict[str, str]:
    time.sleep(1)  # WRONG: blocks the event loop for everybody
    return {'endpoint': 'async-bad'}

سه endpoint که هر کدام یک ثانیه صبر می‌کنند: 01_blocking_vs_async/app.py

load_test.py به هر endpoint ده درخواست هم‌زمان می‌فرستد، درون همین فرایند و بدون سرور:

/async-good  10 requests took 1.0s
/sync-ok     10 requests took 1.1s
/async-bad   10 requests took 10.0s

۲. چند انتظار در یک درخواست

صفحه یک دوره سه چیز مستقل لازم دارد: خود دوره، وضعیت هوای شهر کلاس و نرخ ارز برای قیمت دلاری. سه انتظار مستقل:

@app.get('/courses/{course_id}/slow')
async def page_slow(course_id: int) -> dict:
    started = time.perf_counter()
    course = await load_course(course_id)
    weather = await load_weather('Rasht')
    rate = await load_rate()
    return build(course, weather, rate, started)


@app.get('/courses/{course_id}/fast')
async def page_fast(course_id: int) -> dict:
    started = time.perf_counter()
    course, weather, rate = await asyncio.gather(
        load_course(course_id), load_weather('Rasht'), load_rate()
    )
    return build(course, weather, rate, started)

ترتیبی در برابر gather: 02_concurrent_calls/app.py

02 slow 1202 ms | fast 501 ms

ترتیبی یعنی مجموع انتظارها و gather یعنی طولانی‌ترین آنها. این همان جایی است که async در یک بک‌اند واقعاً سود می‌رساند: صدا زدن سرویس‌های دیگر.

۳. SQLAlchemy ناهمگام

engine = create_async_engine('sqlite+aiosqlite:///async.db')
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(primary_key=True)
    full_name: Mapped[str] = mapped_column(String(120))
    courses: Mapped[list['Course']] = relationship(back_populates='instructor')


class Course(Base):
    __tablename__ = 'courses'
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    instructor_id: Mapped[int] = mapped_column(ForeignKey('users.id'))
    instructor: Mapped[User] = relationship(back_populates='courses')


async def main() -> None:
    async with engine.begin() as connection:
        # create_all is a sync function: run it inside the async connection
        await connection.run_sync(Base.metadata.drop_all)
        await connection.run_sync(Base.metadata.create_all)

    async with SessionLocal() as session:
        ali = User(full_name='Ali Teacher')
        session.add_all(
            [Course(title='Python', instructor=ali), Course(title='FastAPI', instructor=ali)]
        )
        await session.commit()
        print('after commit, still readable:', ali.full_name)

    async with SessionLocal() as session:
        courses = (await session.scalars(select(Course))).all()
        try:
            print(courses[0].instructor.full_name)
        except MissingGreenlet:
            print('lazy loading in async -> MissingGreenlet!')

    async with SessionLocal() as session:
        stmt = select(Course).options(selectinload(Course.instructor)).order_by(Course.id)
        for course in await session.scalars(stmt):
            print(f'{course.title} by {course.instructor.full_name}')

    await engine.dispose()

AsyncSession، MissingGreenlet و selectinload: 03_async_sqlalchemy/app.py

• نشانی به یک درایور ناهمگام اشاره می‌کند: sqlite+aiosqlite:// یا postgresql+asyncpg://.
• هر کاری که ورودی و خروجی دارد await می‌شود: commit، get، scalars، execute، delete، refresh. اما add await نمی‌شود، چون فقط یک شیء را آماده می‌کند.
• create_all یک تابع همگام است؛ درون اتصال ناهمگام با run_sync اجرا می‌شود.

۴. lifespan: یک بار در شروع، یک بار در پایان

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    print('starting: opening one shared HTTP client')
    app.state.http = httpx.AsyncClient(base_url='https://jsonplaceholder.typicode.com', timeout=5)
    app.state.visits = 0
    yield
    print('stopping: closing it')
    await app.state.http.aclose()


app = FastAPI(lifespan=lifespan)


def get_http(request: Request) -> httpx.AsyncClient:
    return request.app.state.http


@app.get('/users/{user_id}/todos')
async def todos(user_id: int, http: Annotated[httpx.AsyncClient, Depends(get_http)]) -> dict:
    try:
        response = await http.get('/todos', params={'userId': user_id})
        response.raise_for_status()
    except httpx.HTTPError as error:
        return {'error': f'upstream API failed: {error!r}'}
    items = response.json()
    return {'user_id': user_id, 'done': sum(t['completed'] for t in items), 'total': len(items)}


@app.get('/visits')
async def visits(request: Request) -> dict[str, int]:
    request.app.state.visits += 1
    return {'visits': request.app.state.visits}

یک کلاینت HTTP مشترک روی app.state: 04_lifespan/app.py

منابع مشترک و گران اینجا ساخته می‌شوند: موتور پایگاه داده (مخزن اتصال‌هایش)، یک httpx.AsyncClient که همه درخواست‌ها از آن استفاده می‌کنند و کش‌ها. آنها را روی app.state می‌گذاریم و از یک وابستگی با request.app به آنها می‌رسیم.

۵. اشتباه‌های رایج، بازسازی‌شده

async def ticker(label: str) -> None:
    """Should print every 0.1 s -- a long gap means the loop was blocked."""
    started = time.perf_counter()
    for _ in range(5):
        print(f'   tick ({label}) at {time.perf_counter() - started:.1f}s')
        await asyncio.sleep(0.1)


async def mistake_1() -> None:
    print('1. forgot await:', fetch_price())  # a coroutine, and a RuntimeWarning later
    print('   with await:  ', await fetch_price())


async def mistake_2_and_4() -> None:
    print('2/4. heavy work directly in async code (ticks stop):')
    started = time.perf_counter()
    await asyncio.gather(ticker('blocked'), asyncio.sleep(0), _run_blocking())
    print(f'   took {time.perf_counter() - started:.2f}s')
    print('     the same work in a thread (ticks continue):')
    started = time.perf_counter()
    await asyncio.gather(ticker('free'), asyncio.to_thread(heavy_hashing))
    print(f'   took {time.perf_counter() - started:.2f}s')


async def _run_blocking() -> None:
    heavy_hashing()  # no await inside: the loop can not switch away

حلقه منجمد و کار سنگین در نخ: 05_common_mistakes/app.py

2/4. heavy work directly in async code (ticks stop):
   tick (blocked) at 0.0s
   tick (blocked) at 0.9s
     the same work in a thread (ticks continue):
   tick (free) at 0.0s
   tick (free) at 0.1s
3. one session shared by 5 tasks -> {'InvalidRequestError'}
   one session per task -> [1, 1, 1, 1, 1]

۶. پروژه: ناهمگام از بالا تا پایین

رفتار API هیچ تغییری نمی‌کند؛ مجموعه Bruno فصل یازدهم روی نسخه ناهمگام دقیقاً همان پاسخ‌ها را می‌گیرد.

settings = get_settings()
# The URL names an ASYNC driver: sqlite+aiosqlite:// or postgresql+asyncpg://
engine = create_async_engine(settings.database_url, echo=settings.database_echo)
# expire_on_commit=False: objects stay readable after commit (no lazy reload in async)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

if engine.dialect.name == 'sqlite':

    @event.listens_for(engine.sync_engine, 'connect')
    def enable_sqlite_foreign_keys(connection, _record) -> None:  # type: ignore[no-untyped-def]
        cursor = connection.cursor()
        cursor.execute('PRAGMA foreign_keys=ON')
        cursor.close()


class Base(DeclarativeBase):
    # Every Mapped[datetime] column keeps the time zone (TIMESTAMP WITH TIME ZONE).
    # Without it PostgreSQL stores "naive" times and asyncpg refuses aware ones.
    type_annotation_map = {datetime: DateTime(timezone=True)}


async def get_db() -> AsyncIterator[AsyncSession]:
    """One AsyncSession per request, closed afterwards."""
    async with SessionLocal() as session:
        yield session

موتور و نشست ناهمگام: 06_project/app/database.py

    async def enroll(self, course_id: int, student_id: int) -> Enrollment:
        course = await CourseService(self.db).get(course_id)
        student = await UserService(self.db).get(student_id)
        if student.role != 'student' or not student.is_active:
            raise BusinessRuleError('Only active students can enroll')
        if not course.is_active:
            raise BusinessRuleError('This course is closed')
        enrollment = await self._find(course_id, student_id)
        if enrollment is not None and enrollment.status == 'active':
            raise ConflictError('Already enrolled')
        if await self.active_count(course_id) >= course.capacity:
            raise ConflictError('Course is full')
        if enrollment is None:
            enrollment = Enrollment(course_id=course_id, student_id=student_id)
            self.db.add(enrollment)
        else:
            enrollment.status = 'active'
        await self.db.commit()
        await self.db.refresh(enrollment)
        return enrollment

همان قاعده ثبت‌نام، حالا با await: 06_project/app/services.py

    async def courses_taught(self, user_id: int) -> list[Course]:
        # `user.courses_taught` would be a lazy load -> not allowed in async: query instead
        await self.get(user_id)
        stmt = select(Course).where(Course.instructor_id == user_id).order_by(Course.title)
        return list(await self.db.scalars(stmt))

صفت رابطه‌ای در async جایش را به یک پرس‌وجو می‌دهد: 06_project/app/services.py

• روترها async def هستند و سرویس‌ها را await می‌کنند.
• ذخیره فایل با await upload.read() و نوشتن روی دیسک از راه anyio انجام می‌شود تا حلقه مسدود نشود.
• lifespan موتور را در پایان می‌بندد. alembic/env.py و seed.py هم ناهمگام شده‌اند.

۷. باگی که PostgreSQL پیدا کرد

مدل‌ها Mapped[datetime] داشتند که TIMESTAMP WITHOUT TIME ZONE می‌سازد، در حالی که کد زمان‌های UTC با منطقه زمانی ذخیره می‌کرد. SQLite اهمیتی نمی‌دهد و درایور همگام psycopg بی‌صدا تبدیل می‌کند. اما asyncpg نمی‌پذیرد:

can't subtract offset-naive and offset-aware datetimes

راه‌حل یک خط در کلاس Base است که حالا در همه پروژه‌ها از فصل پنجم هست:

class Base(DeclarativeBase):
    type_annotation_map = {datetime: DateTime(timezone=True)}

۸. خلاصه فصل