Просунуті залежності¶
🌐 Переклад ШІ та людьми
Цей переклад виконано ШІ під керівництвом людей. 🤝
Можливі помилки через неправильне розуміння початкового змісту або неприродні формулювання тощо. 🤖
Ви можете покращити цей переклад, допомігши нам краще спрямовувати AI LLM.
Параметризовані залежності¶
Усі залежності, які ми бачили, - це фіксована функція або клас.
Але можуть бути випадки, коли ви хочете мати змогу задавати параметри залежності, не оголошуючи багато різних функцій або класів.
Уявімо, що ми хочемо мати залежність, яка перевіряє, чи параметр запиту q містить певний фіксований вміст.
Але ми хочемо мати змогу параметризувати цей фіксований вміст.
Екземпляр «callable»¶
У Python є спосіб зробити екземпляр класу «callable».
Не сам клас (який уже є «callable»), а екземпляр цього класу.
Щоб це зробити, оголошуємо метод __call__:
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
class FixedContentQueryChecker:
def __init__(self, fixed_content: str):
self.fixed_content = fixed_content
def __call__(self, q: str = ""):
if q:
return self.fixed_content in q
return False
checker = FixedContentQueryChecker("bar")
@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
return {"fixed_content_in_query": fixed_content_included}
🤓 Other versions and variants
Tip
Prefer to use the Annotated version if possible.
from fastapi import Depends, FastAPI
app = FastAPI()
class FixedContentQueryChecker:
def __init__(self, fixed_content: str):
self.fixed_content = fixed_content
def __call__(self, q: str = ""):
if q:
return self.fixed_content in q
return False
checker = FixedContentQueryChecker("bar")
@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
return {"fixed_content_in_query": fixed_content_included}
У цьому випадку саме __call__ FastAPI використає для перевірки додаткових параметрів і підзалежностей, і саме його буде викликано, щоб передати значення параметру у вашу функцію операції шляху пізніше.
Параметризувати екземпляр¶
Тепер ми можемо використати __init__, щоб оголосити параметри екземпляра, які можна застосувати для «параметризації» залежності:
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
class FixedContentQueryChecker:
def __init__(self, fixed_content: str):
self.fixed_content = fixed_content
def __call__(self, q: str = ""):
if q:
return self.fixed_content in q
return False
checker = FixedContentQueryChecker("bar")
@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
return {"fixed_content_in_query": fixed_content_included}
🤓 Other versions and variants
Tip
Prefer to use the Annotated version if possible.
from fastapi import Depends, FastAPI
app = FastAPI()
class FixedContentQueryChecker:
def __init__(self, fixed_content: str):
self.fixed_content = fixed_content
def __call__(self, q: str = ""):
if q:
return self.fixed_content in q
return False
checker = FixedContentQueryChecker("bar")
@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
return {"fixed_content_in_query": fixed_content_included}
У цьому випадку FastAPI ніколи не торкається __init__ і не покладається на нього - ми використовуватимемо його безпосередньо у своєму коді.
Створити екземпляр¶
Ми можемо створити екземпляр цього класу так:
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
class FixedContentQueryChecker:
def __init__(self, fixed_content: str):
self.fixed_content = fixed_content
def __call__(self, q: str = ""):
if q:
return self.fixed_content in q
return False
checker = FixedContentQueryChecker("bar")
@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
return {"fixed_content_in_query": fixed_content_included}
🤓 Other versions and variants
Tip
Prefer to use the Annotated version if possible.
from fastapi import Depends, FastAPI
app = FastAPI()
class FixedContentQueryChecker:
def __init__(self, fixed_content: str):
self.fixed_content = fixed_content
def __call__(self, q: str = ""):
if q:
return self.fixed_content in q
return False
checker = FixedContentQueryChecker("bar")
@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
return {"fixed_content_in_query": fixed_content_included}
Таким чином ми «параметризуємо» нашу залежність, яка тепер містить «bar» як атрибут checker.fixed_content.
Використовувати екземпляр як залежність¶
Далі ми можемо використати цей checker у Depends(checker) замість Depends(FixedContentQueryChecker), адже залежністю є екземпляр checker, а не сам клас.
Коли залежність розв'язується, FastAPI викличе цей checker так:
checker(q="somequery")
...і передасть усе, що він поверне, як значення залежності у нашій функції операції шляху як параметр fixed_content_included:
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
class FixedContentQueryChecker:
def __init__(self, fixed_content: str):
self.fixed_content = fixed_content
def __call__(self, q: str = ""):
if q:
return self.fixed_content in q
return False
checker = FixedContentQueryChecker("bar")
@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
return {"fixed_content_in_query": fixed_content_included}
🤓 Other versions and variants
Tip
Prefer to use the Annotated version if possible.
from fastapi import Depends, FastAPI
app = FastAPI()
class FixedContentQueryChecker:
def __init__(self, fixed_content: str):
self.fixed_content = fixed_content
def __call__(self, q: str = ""):
if q:
return self.fixed_content in q
return False
checker = FixedContentQueryChecker("bar")
@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
return {"fixed_content_in_query": fixed_content_included}
Порада
Усе це може здаватися надуманим. І поки що може бути не дуже зрозуміло, навіщо це корисно.
Ці приклади навмисно прості, але показують, як усе працює.
У розділах про безпеку є утилітарні функції, реалізовані таким самим способом.
Якщо ви все це зрозуміли, ви вже знаєте, як під капотом працюють ті утиліти для безпеки.
Залежності з yield, HTTPException, except та фоновими задачами¶
Попередження
Найімовірніше, вам не знадобляться ці технічні деталі.
Вони корисні головно, якщо у вас був застосунок FastAPI старіший за 0.121.0 і ви стикаєтеся з проблемами залежностей із yield.
Залежності з yield еволюціонували з часом, щоб охопити різні сценарії використання та виправити деякі проблеми, ось підсумок змін.
Залежності з yield і scope¶
У версії 0.121.0 FastAPI додано підтримку Depends(scope="function") для залежностей з yield.
З Depends(scope="function") завершальний код після yield виконується одразу після завершення функції операції шляху, до того як відповідь буде надіслана клієнту.
А при використанні Depends(scope="request") (типове значення) завершальний код після yield виконується після відправлення відповіді.
Докладніше читайте в документації: Залежності з yield - Ранній вихід і scope.
Залежності з yield і StreamingResponse, технічні деталі¶
До FastAPI 0.118.0, якщо ви використовували залежність із yield, завершальний код виконувався після повернення з функції операції шляху, але безпосередньо перед відправленням відповіді.
Метою було уникнути утримання ресурсів довше, ніж потрібно, очікуючи, поки відповідь пройде мережею.
Це також означало, що якщо ви повертали StreamingResponse, завершальний код залежності з yield уже було б виконано.
Наприклад, якщо у залежності з yield була сесія бази даних, StreamingResponse не зміг би використовувати цю сесію під час потокової передачі даних, тому що сесію вже закрито в завершальному коді після yield.
Цю поведінку змінено у 0.118.0: завершальний код після yield знову виконується після відправлення відповіді.
Інформація
Як побачите нижче, це дуже схоже на поведінку до версії 0.106.0, але з кількома покращеннями та виправленнями помилок у крайових випадках.
Випадки використання з раннім завершальним кодом¶
Є кілька сценаріїв із певними умовами, які можуть виграти від старої поведінки - виконувати завершальний код залежностей з yield до надсилання відповіді.
Наприклад, уявіть, що у вас є код, який використовує сесію бази даних у залежності з yield лише для перевірки користувача, але сесія більше не використовується у функції операції шляху, тільки в залежності, і відправлення відповіді триває довго - як у StreamingResponse, що повільно надсилає дані, але з якоїсь причини не використовує базу даних.
У такому разі сесія БД утримувалася б до завершення відправлення відповіді, але якщо ви її не використовуєте, утримувати її немає потреби.
Ось як це може виглядати:
import time
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from sqlmodel import Field, Session, SQLModel, create_engine
engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")
class User(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
app = FastAPI()
def get_session():
with Session(engine) as session:
yield session
def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
user = session.get(User, user_id)
if not user:
raise HTTPException(status_code=403, detail="Not authorized")
def generate_stream(query: str):
for ch in query:
yield ch
time.sleep(0.1)
@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
return StreamingResponse(content=generate_stream(query))
Завершальний код - автоматичне закриття Session у:
# Code above omitted 👆
def get_session():
with Session(engine) as session:
yield session
# Code below omitted 👇
👀 Full file preview
import time
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from sqlmodel import Field, Session, SQLModel, create_engine
engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")
class User(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
app = FastAPI()
def get_session():
with Session(engine) as session:
yield session
def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
user = session.get(User, user_id)
if not user:
raise HTTPException(status_code=403, detail="Not authorized")
def generate_stream(query: str):
for ch in query:
yield ch
time.sleep(0.1)
@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
return StreamingResponse(content=generate_stream(query))
...виконається після того, як відповідь завершить надсилати повільні дані:
# Code above omitted 👆
def generate_stream(query: str):
for ch in query:
yield ch
time.sleep(0.1)
@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
return StreamingResponse(content=generate_stream(query))
👀 Full file preview
import time
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from sqlmodel import Field, Session, SQLModel, create_engine
engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")
class User(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
app = FastAPI()
def get_session():
with Session(engine) as session:
yield session
def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
user = session.get(User, user_id)
if not user:
raise HTTPException(status_code=403, detail="Not authorized")
def generate_stream(query: str):
for ch in query:
yield ch
time.sleep(0.1)
@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
return StreamingResponse(content=generate_stream(query))
Але оскільки generate_stream() не використовує сесію бази даних, немає реальної потреби тримати сесію відкритою під час надсилання відповіді.
Якщо у вас саме такий випадок із SQLModel (або SQLAlchemy), ви можете явно закрити сесію, коли вона більше не потрібна:
# Code above omitted 👆
def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
user = session.get(User, user_id)
if not user:
raise HTTPException(status_code=403, detail="Not authorized")
session.close()
# Code below omitted 👇
👀 Full file preview
import time
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from sqlmodel import Field, Session, SQLModel, create_engine
engine = create_engine("postgresql+psycopg://postgres:postgres@localhost/db")
class User(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
app = FastAPI()
def get_session():
with Session(engine) as session:
yield session
def get_user(user_id: int, session: Annotated[Session, Depends(get_session)]):
user = session.get(User, user_id)
if not user:
raise HTTPException(status_code=403, detail="Not authorized")
session.close()
def generate_stream(query: str):
for ch in query:
yield ch
time.sleep(0.1)
@app.get("/generate", dependencies=[Depends(get_user)])
def generate(query: str):
return StreamingResponse(content=generate_stream(query))
Так сесія звільнить з'єднання з базою даних, і його зможуть використовувати інші запити.
Якщо у вас інший сценарій, де потрібно раннє завершення залежності з yield, створіть, будь ласка, питання в обговореннях GitHub із вашим конкретним випадком і поясненням, чому вам корисне раннє закриття для залежностей з yield.
Якщо будуть переконливі приклади для раннього закриття в залежностях з yield, я розгляну додавання нового способу увімкнути раннє закриття.
Залежності з yield і except, технічні деталі¶
До FastAPI 0.110.0, якщо ви використовували залежність із yield, перехоплювали виняток через except у цій залежності і не піднімали його знову, виняток автоматично піднімався/пересилався до будь-яких обробників винятків або внутрішнього обробника помилок сервера.
Це змінено у версії 0.110.0, щоб усунути неконтрольоване споживання пам'яті від пересланих винятків без обробника (внутрішні помилки сервера) та зробити поведінку узгодженою зі звичайним Python-кодом.
Фонові задачі та залежності з yield, технічні деталі¶
До FastAPI 0.106.0 піднімати винятки після yield було неможливо: завершальний код у залежностях з yield виконувався після надсилання відповіді, тож обробники винятків уже відпрацювали б.
Так було спроєктовано головно для того, щоб дозволити використовувати ті самі об'єкти, «віддані» залежностями через yield, усередині фонових задач, оскільки завершальний код виконувався після завершення фонових задач.
У FastAPI 0.106.0 це змінено, щоб не утримувати ресурси під час очікування, поки відповідь піде мережею.
Порада
Крім того, фонова задача зазвичай є незалежним набором логіки, який слід обробляти окремо, з власними ресурсами (наприклад, власним з'єднанням з базою даних).
Тож так у вас, ймовірно, буде чистіший код.
Якщо ви раніше покладалися на цю поведінку, тепер слід створювати ресурси для фонових задач усередині самої фонової задачі та використовувати всередині лише дані, що не залежать від ресурсів залежностей із yield.
Наприклад, замість використання тієї самої сесії бази даних ви створюватимете нову сесію в самій фоновій задачі та отримуватимете об'єкти з бази даних, використовуючи цю нову сесію. І далі, замість передавання об'єкта з бази даних як параметра у функцію фонової задачі, ви передасте ідентифікатор цього об'єкта, а потім отримаєте об'єкт знову всередині функції фонової задачі.