Внедрение зависимостей, переиспользование кода, Depends
Dependency Injection — мощный паттерн для переиспользования кода, разделения ответственности и чистой архитектуры. В этой теме вы научитесь использовать
Depends()для внедрения зависимостей.
Dependency Injection (внедрение зависимостей) — паттерн, при котором зависимости (сервисы, репозитории, конфигурация) передаются в объект извне, а не создаются внутри.
@app.get('/users/{user_id}')
def get_user(user_id: int):
db = SessionLocal() # Создаём зависимость внутри
try:
user = db.query(User).filter(User.id == user_id).first()
return user
finally:
db.close()Проблемы:
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get('/users/{user_id}')
def get_user(user_id: int, db: Session = Depends(get_db)):
user = db.query(User).filter(User.id == user_id).first()
return userПреимущества:
get_db)Depends() говорит FastAPI: «Получи эту зависимость и передай в функцию».
from fastapi import FastAPI, Depends
app = FastAPI()
def get_common_params(q: str | None = None, skip: int = 0, limit: int = 100):
return {'q': q, 'skip': skip, 'limit': limit}
@app.get('/items')
def list_items(params: dict = Depends(get_common_params)):
return paramsFastAPI:
get_common_params() с query-параметрамиlist_items как параметр paramsДля ресурсов, требующих очистки (БД, соединения), используется yield:
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
def get_db():
db = SessionLocal()
try:
yield db # Передаём db в эндпоинт
finally:
db.close() # Закрываем после выполнения
@app.get('/users', response_model=list[UserResponse])
def list_users(db: Session = Depends(get_db)):
return db.query(User).all()1. Запрос приходит
↓
2. FastAPI вызывает get_db()
↓
3. get_db() создаёт db, делает yield db
↓
4. FastAPI передаёт db в list_users()
↓
5. list_users() выполняется, возвращает ответ
↓
6. FastAPI возобновляет get_db() после yield
↓
7. get_db() закрывает соединение (finally)
↓
8. Ответ отправляется клиенту
Зависимости могут использовать другие зависимости:
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
def get_current_user(db: Session = Depends(get_db)):
# Получаем токен из заголовков
token = get_token_from_headers()
user = db.query(User).filter(User.token == token).first()
if not user:
raise HTTPException(status_code=401, detail="Not authenticated")
return user
def get_current_active_user(
current_user: User = Depends(get_current_user)
):
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
@app.get('/users/me')
def read_current_user(
user: User = Depends(get_current_active_user)
):
return userread_current_user
↓ Depends(get_current_active_user)
get_current_active_user
↓ Depends(get_current_user)
get_current_user
↓ Depends(get_db)
get_db (yield db)
FastAPI разрешает зависимости в правильном порядке.
Можно использовать классы вместо функций:
class Pagination:
def __init__(self, page: int = 1, page_size: int = 10):
self.page = page
self.page_size = page_size
self.skip = (page - 1) * page_size
@app.get('/items')
def list_items(pagination: Pagination = Depends()):
items = db.query(Item).offset(pagination.skip).limit(pagination.page_size).all()
return {'items': items, 'page': pagination.page}FastAPI создаст экземпляр Pagination, передав query-параметры в __init__.
Классический случай DI — проверка аутентификации:
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
SECRET_KEY = "secret"
ALGORITHM = "HS256"
def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = get_user_from_db(username)
if user is None:
raise credentials_exception
return user
@app.get('/users/me')
def read_current_user(current_user: User = Depends(get_current_user)):
return current_user@app.get('/users/me')
def read_current_user(current_user: User = Depends(get_current_user)):
return current_user
@app.put('/users/me')
def update_current_user(
user_update: UserUpdate,
current_user: User = Depends(get_current_user)
):
return update_user_in_db(current_user.id, user_update)
@app.delete('/users/me', status_code=204)
def delete_current_user(
current_user: User = Depends(get_current_user)
):
delete_user_from_db(current_user.id)Можно передавать параметры в зависимости:
def get_pagination(page: int = 1, page_size: int = 10):
skip = (page - 1) * page_size
return {'skip': skip, 'limit': page_size}
@app.get('/items')
def list_items(pagination: dict = Depends(get_pagination)):
items = db.query(Item).offset(pagination['skip']).limit(pagination['limit']).all()
return itemsdef get_filter(model_class: type, field: str, value: str):
def filter_func(db: Session = Depends(get_db)):
return db.query(model_class).filter(getattr(model_class, field) == value).first()
return filter_func
@app.get('/users/{username}')
def get_user(user: User = Depends(lambda: get_filter(User, 'username', ...))):
return userМожно применить зависимость ко всему приложению:
from fastapi import FastAPI, Depends
app = FastAPI(dependencies=[Depends(verify_token)])
async def verify_token():
# Проверка токена для всех эндпоинтов
token = get_token()
if not is_valid(token):
raise HTTPException(status_code=401, detail="Invalid token")
@app.get('/items')
def list_items():
return db.query(Item).all() # Токен уже проверен
@app.get('/users')
def list_users():
return db.query(User).all() # Токен уже проверенМожно применить зависимость к конкретному роутеру:
from fastapi import APIRouter, Depends
router = APIRouter(dependencies=[Depends(get_current_user)])
@router.get('/items')
def list_items():
...
@router.get('/users')
def list_users():
...
app.include_router(router, prefix='/api')Все эндпоинты в роутере требуют аутентификации.
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import Annotated
app = FastAPI()
# === База данных ===
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
DbDependency = Annotated[Session, Depends(get_db)]
# === Аутентификация ===
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
def get_current_user(
token: Annotated[str, Depends(oauth2_scheme)],
db: DbDependency
):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = db.query(User).filter(User.username == username).first()
if user is None:
raise credentials_exception
return user
UserDependency = Annotated[User, Depends(get_current_user)]
def get_current_active_user(current_user: UserDependency):
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
ActiveUserDependency = Annotated[User, Depends(get_current_active_user)]
def get_current_admin_user(current_user: UserDependency):
if not current_user.is_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions"
)
return current_user
AdminUserDependency = Annotated[User, Depends(get_current_admin_user)]
# === Пагинация ===
def get_pagination(page: int = 1, page_size: int = 20):
skip = (page - 1) * page_size
return {'skip': skip, 'limit': page_size}
PaginationDependency = Annotated[dict, Depends(get_pagination)]
# === Endpoints ===
@app.get('/users/me')
def read_current_user(current_user: ActiveUserDependency):
return current_user
@app.get('/users')
def list_users(
current_user: AdminUserDependency,
pagination: PaginationDependency
):
users = db.query(User).offset(pagination['skip']).limit(pagination['limit']).all()
return users
@app.post('/posts')
def create_post(
post: PostCreate,
current_user: ActiveUserDependency,
db: DbDependency
):
new_post = Post(**post.model_dump(), author_id=current_user.id)
db.add(new_post)
db.commit()
db.refresh(new_post)
return new_post
@app.delete('/posts/{post_id}')
def delete_post(
post_id: int,
current_user: AdminUserDependency,
db: DbDependency
):
post = db.query(Post).filter(Post.id == post_id).first()
if not post:
raise HTTPException(status_code=404, detail="Post not found")
db.delete(post)
db.commit()
return {'deleted': True}DI позволяет легко подменять зависимости в тестах:
from fastapi.testclient import TestClient
# Тестовая БД
def override_get_db():
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)
def test_read_users():
response = client.get("/users")
assert response.status_code == 200@app.get('/items')
def list_items(db: Session = Depends(get_db)): # Забыли ()
...Проблема: Depends(get_db) вместо Depends(get_db()) — неправильно, но работает, потому что FastAPI понимает оба варианта.
Решение: Используйте Depends(get_db) — это предпочтительный синтаксис.
def get_db():
db = SessionLocal()
yield db
# Забыли db.close()!Проблема: Соединение не закроется при ошибке.
Решение:
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()def dep_a(dep_b = Depends(get_dep_b)): ...
def dep_b(dep_a = Depends(get_dep_a)): ...Проблема: Бесконечный цикл.
Решение: Перестройте зависимости, чтобы не было циклов.
__init__dependency_overridesВ следующей теме вы изучите работу с базой данных — SQLAlchemy, асинхронные ORM, миграции Alembic.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.