Залежності з yield

🌐 Переклад ШІ та людьми

Цей переклад виконано ШІ під керівництвом людей. 🤝

Можливі помилки через неправильне розуміння початкового змісту або неприродні формулювання тощо. 🤖

Ви можете покращити цей переклад, допомігши нам краще спрямовувати AI LLM.

Англійська версія

FastAPI підтримує залежності, які виконують деякі додаткові кроки після завершення.

Щоб це зробити, використовуйте yield замість return і напишіть додаткові кроки (код) після нього.

Порада

Переконайтесь, що ви використовуєте yield лише один раз на залежність.

Технічні деталі

Будь-яка функція, яку можна використовувати з:

• @contextlib.contextmanager або
• @contextlib.asynccontextmanager

буде придатною як залежність у FastAPI.

Насправді FastAPI використовує ці два декоратори внутрішньо.

Залежність бази даних з yield

Наприклад, ви можете використати це, щоб створити сесію бази даних і закрити її після завершення.

Перед створенням відповіді виконується лише код до і включно з оператором yield:

Python 3.10+

async def get_db():
    db = DBSession()
    try:
        yield db
    finally:
        db.close()

Значення, передане yield, впроваджується в операції шляху та інші залежності:

Python 3.10+

async def get_db():
    db = DBSession()
    try:
        yield db
    finally:
        db.close()

Код після оператора yield виконується після відповіді:

Python 3.10+

async def get_db():
    db = DBSession()
    try:
        yield db
    finally:
        db.close()

Порада

Можете використовувати як async, так і звичайні функції.

FastAPI зробить усе правильно з кожною з них, так само як і зі звичайними залежностями.

Залежність з yield та try

Якщо ви використовуєте блок try в залежності з yield, ви отримаєте будь-який виняток, який був згенерований під час використання залежності.

Наприклад, якщо якийсь код десь посередині, в іншій залежності або в операції шляху, зробив «rollback» транзакції бази даних або створив будь-який інший виняток, ви отримаєте цей виняток у своїй залежності.

Тож ви можете обробити цей конкретний виняток усередині залежності за допомогою except SomeException.

Так само ви можете використовувати finally, щоб гарантувати виконання завершальних кроків незалежно від того, був виняток чи ні.

Python 3.10+

async def get_db():
    db = DBSession()
    try:
        yield db
    finally:
        db.close()

Підзалежності з yield

Ви можете мати підзалежності та «дерева» підзалежностей будь-якого розміру і форми, і будь-яка або всі з них можуть використовувати yield.

FastAPI гарантує, що «exit code» у кожній залежності з yield буде виконано в правильному порядку.

Наприклад, dependency_c може залежати від dependency_b, а dependency_b - від dependency_a:

Python 3.10+

from typing import Annotated

from fastapi import Depends


async def dependency_a():
    dep_a = generate_dep_a()
    try:
        yield dep_a
    finally:
        dep_a.close()


async def dependency_b(dep_a: Annotated[DepA, Depends(dependency_a)]):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)


async def dependency_c(dep_b: Annotated[DepB, Depends(dependency_b)]):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)

🤓 Other versions and variants

Python 3.10+ - non-Annotated

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends


async def dependency_a():
    dep_a = generate_dep_a()
    try:
        yield dep_a
    finally:
        dep_a.close()


async def dependency_b(dep_a=Depends(dependency_a)):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)


async def dependency_c(dep_b=Depends(dependency_b)):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)

І всі вони можуть використовувати yield.

У цьому випадку dependency_c, щоб виконати свій завершальний код, потребує, щоб значення з dependency_b (тут dep_b) все ще було доступним.

І, у свою чергу, dependency_b потребує, щоб значення з dependency_a (тут dep_a) було доступним для свого завершального коду.

Python 3.10+

from typing import Annotated

from fastapi import Depends


async def dependency_a():
    dep_a = generate_dep_a()
    try:
        yield dep_a
    finally:
        dep_a.close()


async def dependency_b(dep_a: Annotated[DepA, Depends(dependency_a)]):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)


async def dependency_c(dep_b: Annotated[DepB, Depends(dependency_b)]):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)

🤓 Other versions and variants

Python 3.10+ - non-Annotated

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends


async def dependency_a():
    dep_a = generate_dep_a()
    try:
        yield dep_a
    finally:
        dep_a.close()


async def dependency_b(dep_a=Depends(dependency_a)):
    dep_b = generate_dep_b()
    try:
        yield dep_b
    finally:
        dep_b.close(dep_a)


async def dependency_c(dep_b=Depends(dependency_b)):
    dep_c = generate_dep_c()
    try:
        yield dep_c
    finally:
        dep_c.close(dep_b)

Так само ви можете мати деякі залежності з yield, а інші - з return, і частина з них може залежати від інших.

І ви можете мати одну залежність, яка вимагає кілька інших залежностей з yield тощо.

Ви можете мати будь-які комбінації залежностей, які вам потрібні.

FastAPI подбає, щоб усе виконувалося в правильному порядку.

Технічні деталі

Це працює завдяки Менеджерам контексту Python.

FastAPI використовує їх внутрішньо, щоб досягти цього.

Залежності з yield та HTTPException

Ви бачили, що можна використовувати залежності з yield і мати блоки try, які намагаються виконати деякий код, а потім запускають завершальний код після finally.

Також можна використовувати except, щоб перехопити згенерований виняток і щось із ним зробити.

Наприклад, ви можете підняти інший виняток, як-от HTTPException.

Порада

Це доволі просунута техніка, і в більшості випадків вона вам не знадобиться, адже ви можете піднімати винятки (включно з HTTPException) всередині іншого коду вашого застосунку, наприклад, у функції операції шляху.

Але вона є, якщо вам це потрібно. 🤓

Python 3.10+

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


data = {
    "plumbus": {"description": "Freshly pickled plumbus", "owner": "Morty"},
    "portal-gun": {"description": "Gun to create portals", "owner": "Rick"},
}


class OwnerError(Exception):
    pass


def get_username():
    try:
        yield "Rick"
    except OwnerError as e:
        raise HTTPException(status_code=400, detail=f"Owner error: {e}")


@app.get("/items/{item_id}")
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]):
    if item_id not in data:
        raise HTTPException(status_code=404, detail="Item not found")
    item = data[item_id]
    if item["owner"] != username:
        raise OwnerError(username)
    return item

🤓 Other versions and variants

Python 3.10+ - non-Annotated

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


data = {
    "plumbus": {"description": "Freshly pickled plumbus", "owner": "Morty"},
    "portal-gun": {"description": "Gun to create portals", "owner": "Rick"},
}


class OwnerError(Exception):
    pass


def get_username():
    try:
        yield "Rick"
    except OwnerError as e:
        raise HTTPException(status_code=400, detail=f"Owner error: {e}")


@app.get("/items/{item_id}")
def get_item(item_id: str, username: str = Depends(get_username)):
    if item_id not in data:
        raise HTTPException(status_code=404, detail="Item not found")
    item = data[item_id]
    if item["owner"] != username:
        raise OwnerError(username)
    return item

Якщо ви хочете перехоплювати винятки та створювати на їх основі користувацьку відповідь, створіть Користувацький обробник винятків.

Залежності з yield та except

Якщо ви перехоплюєте виняток за допомогою except у залежності з yield і не піднімаєте його знову (або не піднімаєте новий виняток), FastAPI не зможе помітити, що стався виняток, так само як це було б у звичайному Python:

Python 3.10+

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


class InternalError(Exception):
    pass


def get_username():
    try:
        yield "Rick"
    except InternalError:
        print("Oops, we didn't raise again, Britney 😱")


@app.get("/items/{item_id}")
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]):
    if item_id == "portal-gun":
        raise InternalError(
            f"The portal gun is too dangerous to be owned by {username}"
        )
    if item_id != "plumbus":
        raise HTTPException(
            status_code=404, detail="Item not found, there's only a plumbus here"
        )
    return item_id

🤓 Other versions and variants

Python 3.10+ - non-Annotated

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


class InternalError(Exception):
    pass


def get_username():
    try:
        yield "Rick"
    except InternalError:
        print("Oops, we didn't raise again, Britney 😱")


@app.get("/items/{item_id}")
def get_item(item_id: str, username: str = Depends(get_username)):
    if item_id == "portal-gun":
        raise InternalError(
            f"The portal gun is too dangerous to be owned by {username}"
        )
    if item_id != "plumbus":
        raise HTTPException(
            status_code=404, detail="Item not found, there's only a plumbus here"
        )
    return item_id

У цьому випадку клієнт побачить відповідь HTTP 500 Internal Server Error, як і має бути, з огляду на те, що ми не піднімаємо HTTPException або подібний виняток, але на сервері не буде жодних логів чи інших ознак того, що це була за помилка. 😱

Завжди використовуйте raise у залежностях з yield та except

Якщо ви перехоплюєте виняток у залежності з yield, якщо тільки ви не піднімаєте інший HTTPException або подібний, вам слід повторно підняти початковий виняток.

Ви можете повторно підняти той самий виняток, використовуючи raise:

Python 3.10+

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


class InternalError(Exception):
    pass


def get_username():
    try:
        yield "Rick"
    except InternalError:
        print("We don't swallow the internal error here, we raise again 😎")
        raise


@app.get("/items/{item_id}")
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]):
    if item_id == "portal-gun":
        raise InternalError(
            f"The portal gun is too dangerous to be owned by {username}"
        )
    if item_id != "plumbus":
        raise HTTPException(
            status_code=404, detail="Item not found, there's only a plumbus here"
        )
    return item_id

🤓 Other versions and variants

Python 3.10+ - non-Annotated

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()


class InternalError(Exception):
    pass


def get_username():
    try:
        yield "Rick"
    except InternalError:
        print("We don't swallow the internal error here, we raise again 😎")
        raise


@app.get("/items/{item_id}")
def get_item(item_id: str, username: str = Depends(get_username)):
    if item_id == "portal-gun":
        raise InternalError(
            f"The portal gun is too dangerous to be owned by {username}"
        )
    if item_id != "plumbus":
        raise HTTPException(
            status_code=404, detail="Item not found, there's only a plumbus here"
        )
    return item_id

Тепер клієнт отримає ту саму відповідь HTTP 500 Internal Server Error, але сервер матиме наш користувацький InternalError у логах. 😎

Виконання залежностей з yield

Послідовність виконання приблизно така, як на цій діаграмі. Час тече зверху вниз. І кожна колонка - це одна з частин, що взаємодіють або виконують код.

sequenceDiagram

participant client as Client
participant handler as Exception handler
participant dep as Dep with yield
participant operation as Path Operation
participant tasks as Background tasks

    Note over client,operation: Can raise exceptions, including HTTPException
    client ->> dep: Start request
    Note over dep: Run code up to yield
    opt raise Exception
        dep -->> handler: Raise Exception
        handler -->> client: HTTP error response
    end
    dep ->> operation: Run dependency, e.g. DB session
    opt raise
        operation -->> dep: Raise Exception (e.g. HTTPException)
        opt handle
            dep -->> dep: Can catch exception, raise a new HTTPException, raise other exception
        end
        handler -->> client: HTTP error response
    end

    operation ->> client: Return response to client
    Note over client,operation: Response is already sent, can't change it anymore
    opt Tasks
        operation -->> tasks: Send background tasks
    end
    opt Raise other exception
        tasks -->> tasks: Handle exceptions in the background task code
    end

Інформація

Лише одна відповідь буде надіслана клієнту. Це може бути одна з помилкових відповідей або відповідь від операції шляху.

Після відправлення однієї з цих відповідей іншу відправити не можна.

Порада

Якщо ви піднімаєте будь-який виняток у коді з функції операції шляху, він буде переданий у залежності з yield, включно з HTTPException. У більшості випадків ви захочете повторно підняти той самий виняток або новий із залежності з yield, щоб переконатися, що його коректно оброблено.

Ранній вихід і scope

Зазвичай завершальний код залежностей з yield виконується після того, як відповідь надіслано клієнту.

Але якщо ви знаєте, що вам не потрібно використовувати залежність після повернення з функції операції шляху, ви можете використати Depends(scope="function"), щоб сказати FastAPI, що слід закрити залежність після повернення з функції операції шляху, але до надсилання відповіді.

Python 3.10+

from typing import Annotated

from fastapi import Depends, FastAPI

app = FastAPI()


def get_username():
    try:
        yield "Rick"
    finally:
        print("Cleanup up before response is sent")


@app.get("/users/me")
def get_user_me(username: Annotated[str, Depends(get_username, scope="function")]):
    return username

🤓 Other versions and variants

Python 3.10+ - non-Annotated

Tip

Prefer to use the Annotated version if possible.

from fastapi import Depends, FastAPI

app = FastAPI()


def get_username():
    try:
        yield "Rick"
    finally:
        print("Cleanup up before response is sent")


@app.get("/users/me")
def get_user_me(username: str = Depends(get_username, scope="function")):
    return username

Depends() приймає параметр scope, який може бути:

• "function": запустити залежність перед функцією операції шляху, що обробляє запит, завершити залежність після завершення функції операції шляху, але до того, як відповідь буде відправлена клієнту. Тобто функція залежності буде виконуватися навколо функції операції шляху.
• "request": запустити залежність перед функцією операції шляху, що обробляє запит (подібно до "function"), але завершити після того, як відповідь буде відправлена клієнту. Тобто функція залежності буде виконуватися навколо циклу запиту та відповіді.

Якщо не вказано, і залежність має yield, за замовчуванням scope дорівнює "request".

scope для підзалежностей

Коли ви оголошуєте залежність із scope="request" (за замовчуванням), будь-яка підзалежність також має мати scope рівний "request".

Але залежність з scope рівним "function" може мати залежності з scope "function" і scope "request".

Це тому, що будь-яка залежність має бути здатною виконати свій завершальний код раніше за підзалежності, оскільки вона може все ще потребувати їх під час свого завершального коду.

sequenceDiagram

participant client as Client
participant dep_req as Dep scope="request"
participant dep_func as Dep scope="function"
participant operation as Path Operation

    client ->> dep_req: Start request
    Note over dep_req: Run code up to yield
    dep_req ->> dep_func: Pass dependency
    Note over dep_func: Run code up to yield
    dep_func ->> operation: Run path operation with dependency
    operation ->> dep_func: Return from path operation
    Note over dep_func: Run code after yield
    Note over dep_func: ✅ Dependency closed
    dep_func ->> client: Send response to client
    Note over client: Response sent
    Note over dep_req: Run code after yield
    Note over dep_req: ✅ Dependency closed

Залежності з yield, HTTPException, except і фоновими задачами

Залежності з yield еволюціонували з часом, щоб покрити різні сценарії та виправити деякі проблеми.

Якщо ви хочете дізнатися, що змінювалося в різних версіях FastAPI, прочитайте про це в просунутому посібнику користувача: Розширені залежності - Залежності з yield, HTTPException, except і фоновими задачами.

Менеджери контексту

Що таке «Менеджери контексту»

«Менеджери контексту» - це будь-які Python-об'єкти, які можна використовувати в операторі with.

Наприклад, можна використати with, щоб прочитати файл:

with open("./somefile.txt") as f:
    contents = f.read()
    print(contents)

Під капотом open("./somefile.txt") створює об'єкт, який називається «Менеджер контексту».

Коли блок with завершується, він гарантує закриття файлу, навіть якщо були винятки.

Коли ви створюєте залежність з yield, FastAPI внутрішньо створить для неї менеджер контексту й поєднає його з іншими пов'язаними інструментами.

Використання менеджерів контексту в залежностях з yield

Попередження

Це, загалом, «просунута» ідея.

Якщо ви тільки починаєте з FastAPI, можливо, варто наразі пропустити це.

У Python ви можете створювати Менеджери контексту, створивши клас із двома методами: __enter__() і __exit__().

Ви також можете використовувати їх усередині залежностей FastAPI з yield, використовуючи with або async with у середині функції залежності:

Python 3.10+

class MySuperContextManager:
    def __init__(self):
        self.db = DBSession()

    def __enter__(self):
        return self.db

    def __exit__(self, exc_type, exc_value, traceback):
        self.db.close()


async def get_db():
    with MySuperContextManager() as db:
        yield db

Порада

Інший спосіб створити менеджер контексту:

• @contextlib.contextmanager або
• @contextlib.asynccontextmanager

використовуючи їх для декорування функції з одним yield.

Саме це FastAPI використовує внутрішньо для залежностей з yield.

Але вам не потрібно використовувати ці декоратори для залежностей FastAPI (і не варто).

FastAPI зробить це за вас внутрішньо.