Интеграция двух фреймворков: FastAPI внутри Django и наоборот

Расширенные примеры интеграции с кодом и результатами

Ниже приведены полные сценарии: от настройки Django ORM в FastAPI до совместной аутентификации.

Пример 1. Полноценное приложение FastAPI с моделями Django и миграциями

# 1. Структура проекта
myproject/
├── myproject/
│   ├── __init__.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
├── myapp/
│   ├── __init__.py
│   ├── models.py
│   ├── admin.py
│   └── migrations/
├── fastapi_app/
│   ├── __init__.py
│   ├── main.py
│   └── dependencies.py
├── manage.py
└── requirements.txt

# 2. settings.py - настройки БД и установка приложения
INSTALLED_APPS = [
    'django.contrib.contenttypes',
    'django.contrib.auth',
    'myapp',
]
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'integrated_db',
        'USER': 'admin',
        'PASSWORD': 'secret',
        'HOST': 'localhost',
        'PORT': '5432',
    }
}

# 3. models.py (Django) - модель с пользовательскими полями
from django.db import models

class Article(models.Model):
    title = models.CharField(max_length=200)
    content = models.TextField()
    published = models.BooleanField(default=False)
    created_at = models.DateTimeField(auto_now_add=True)

    class Meta:
        db_table = 'articles'

# 4. Миграции
# python manage.py makemigrations myapp
# python manage.py migrate

# 5. fastapi_app/main.py - FastAPI с Django ORM
import os
import django
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from asgiref.sync import sync_to_async
from datetime import datetime

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
django.setup()

from myapp.models import Article

app = FastAPI(title='Articles API')

class ArticleIn(BaseModel):
    title: str
    content: str
    published: bool = False

class ArticleOut(BaseModel):
    id: int
    title: str
    content: str
    published: bool
    created_at: datetime

@sync_to_async
def create_article(data):
    article = Article.objects.create(
        title=data.title,
        content=data.content,
        published=data.published
    )
    return article

@sync_to_async
def list_articles():
    return list(Article.objects.all().values('id', 'title', 'content', 'published', 'created_at'))

@app.post("/articles", response_model=ArticleOut)
async def add_article(article: ArticleIn):
    obj = await create_article(article)
    return ArticleOut(
        id=obj.id,
        title=obj.title,
        content=obj.content,
        published=obj.published,
        created_at=obj.created_at
    )

@app.get("/articles")
async def get_articles():
    items = await list_articles()
    return items

# Результат запроса POST /articles
# Body: { "title": "Test", "content": "Some text", "published": true }
# Ответ:
{
  "id": 1,
  "title": "Test",
  "content": "Some text",
  "published": true,
  "created_at": "2025-03-25T10:00:00Z"
}

# GET /articles
[
  {
    "id": 1,
    "title": "Test",
    "content": "Some text",
    "published": true,
    "created_at": "2025-03-25T10:00:00Z"
  }
]

Пример 2. Использование Django аутентификации в FastAPI (сессии или JWT)

При запуске Django ORM внутри FastAPI можно использовать встроенную систему аутентификации. Например, проверять пользователя по сессионной куке или по токену из заголовка.

# dependencies.py
from fastapi import Header, HTTPException
from asgiref.sync import sync_to_async
from django.contrib.auth import get_user_model
from django.contrib.auth.models import AnonymousUser

User = get_user_model()

@sync_to_async
def get_user_by_token(token: str):
    # Пример: ищем пользователя по токену в модели Token (django-rest-framework)
    from rest_framework.authtoken.models import Token
    try:
        token_obj = Token.objects.get(key=token)
        return token_obj.user
    except Token.DoesNotExist:
        return AnonymousUser()

async def get_current_user(authorization: str = Header(None)):
    if not authorization or not authorization.startswith('Token '):
        raise HTTPException(status_code=401, detail="Missing or invalid token")
    token = authorization.split(' ')[1]
    user = await get_user_by_token(token)
    if user.is_anonymous:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

# main.py - защищенный эндпоинт
from fastapi import Depends
from .dependencies import get_current_user

@app.get("/me")
async def read_current_user(user: User = Depends(get_current_user)):
    return {
        "id": user.id,
        "username": user.username,
        "email": user.email
    }

Пример 3. Совместная миграция данных при общих моделях

Если используется подход с двумя отдельными приложениями (Django и FastAPI) и общей базой, то для применения миграций достаточно запускать python manage.py migrate в проекте Django. FastAPI при этом должен работать с той же схемой. Рекомендуется использовать общий пакет с моделями, установленный через pip.

# shared_models/
#   __init__.py
#   models.py (содержит те же классы, что и в Django)
#   setups.py (функция инициализации Django settings)

# В проекте FastAPI:
from shared_models.setups import init_django
init_django()  # вызывает django.setup()
from shared_models.models import Product