committed by
GitHub
13 changed files with 274 additions and 0 deletions
@ -0,0 +1,91 @@ |
|||
# Using the Django ORM with FastAPI |
|||
|
|||
In this guide we'll show you how to use Django's ORM with FastAPI. |
|||
|
|||
This can be extremely useful when migrating from Django to FastAPI, as you can reuse your existing Django models and queries. It's also a great way to take advantage of Django's powerful ORM while using FastAPI's modern features. |
|||
|
|||
This tutorial is based on the Django polls tutorial, but you can apply the same concepts to any Django project. |
|||
|
|||
## Prerequisites |
|||
|
|||
- A Django project like the one created from the [Django polls tutorial](https://docs.djangoproject.com/en/stable/intro/tutorial01/) |
|||
- Basic knowledge of FastAPI |
|||
|
|||
## Step 1: Install FastAPI |
|||
|
|||
First, let's install FastAPI in our Django project: |
|||
|
|||
```bash |
|||
# make sure to run this in your Django virtual environment |
|||
pip install fastapi |
|||
``` |
|||
|
|||
## Step 2: Set up a basic FastAPI application |
|||
|
|||
Let's create a basic FastAPI application in a new file called `main.py`: |
|||
|
|||
//// tab | Python 3.8+ |
|||
|
|||
```Python |
|||
{!> ../../../docs_src/django_orm/tutorial001.py[ln:5,13]!} |
|||
``` |
|||
|
|||
In the next steps we'll import the `Question` model from Django, and create a FastAPI endpoint to list all questions. |
|||
|
|||
## Step 3: Import and use Django models |
|||
|
|||
In your `main.py` file, let's import the `Question` model and create a FastAPI endpoint to list all questions: |
|||
|
|||
//// tab | Python 3.8+ |
|||
|
|||
```Python |
|||
{!> ../../../docs_src/django_orm/tutorial001.py[ln:10,16-20]!} |
|||
``` |
|||
|
|||
## Step 4: Run the FastAPI application |
|||
|
|||
Now, let's run the FastAPI application: |
|||
|
|||
```bash |
|||
fastapi dev main.py |
|||
``` |
|||
|
|||
If you go to `http://localhost:8000/questions` we should see the list of questions, right? 🤔 |
|||
|
|||
Unfortunately, we'll get this error: |
|||
|
|||
```text |
|||
django.core.exceptions.ImproperlyConfigured: Requested setting INSTALLED_APPS, but settings are not configured. |
|||
You must either define the environment variable DJANGO_SETTINGS_MODULE or call settings.configure() before accessing settings. |
|||
``` |
|||
|
|||
This error happens because Django needs to be configured before importing the models. |
|||
|
|||
## Step 5: Configure Django settings |
|||
|
|||
To fix this error, we need to configure Django settings before importing the Django models. |
|||
|
|||
In the `main.py` add the following code **before** importing the Django models: |
|||
|
|||
//// tab | Python 3.8+ |
|||
|
|||
```Python |
|||
{!> ../../../docs_src/django_orm/tutorial001.py[ln:1,7-10]!} |
|||
``` |
|||
|
|||
Now, if you run the FastAPI application again, you should see the list of questions at `http://localhost:8000/questions`! 🎉 |
|||
|
|||
## Conclusion |
|||
|
|||
In this guide, we learned how to use Django's ORM with FastAPI. This can be extremely useful when migrating from Django to FastAPI, as you can reuse your existing Django models and queries. |
|||
|
|||
## Using the ORM in async routes |
|||
|
|||
Django's support for async is currently limited, if you need to do run any query in an async route (or function), |
|||
you need to either use the async equivalent of the query or use `sync_to_async` from `asgiref.sync` to run the query: |
|||
|
|||
//// tab | Python 3.8+ |
|||
|
|||
```Python |
|||
{!> ../../../docs_src/django_orm/tutorial001.py[ln:23-37]!} |
|||
``` |
@ -0,0 +1,11 @@ |
|||
INSTALLED_APPS = ["polls"] |
|||
|
|||
ROOT_URLCONF = "mysite.urls" |
|||
USE_TZ = False |
|||
|
|||
DATABASES = { |
|||
"default": { |
|||
"ENGINE": "django.db.backends.sqlite3", |
|||
"NAME": ":memory:", |
|||
}, |
|||
} |
@ -0,0 +1 @@ |
|||
urlpatterns = [] |
@ -0,0 +1,52 @@ |
|||
# Generated by Django 2.1 on 2018-08-08 21:45 |
|||
|
|||
import django.db.models.deletion |
|||
from django.db import migrations, models |
|||
|
|||
|
|||
class Migration(migrations.Migration): |
|||
initial = True |
|||
|
|||
dependencies = [] |
|||
|
|||
operations = [ |
|||
migrations.CreateModel( |
|||
name="Choice", |
|||
fields=[ |
|||
( |
|||
"id", |
|||
models.AutoField( |
|||
auto_created=True, |
|||
primary_key=True, |
|||
serialize=False, |
|||
verbose_name="ID", |
|||
), |
|||
), |
|||
("choice_text", models.CharField(max_length=200)), |
|||
("votes", models.IntegerField(default=0)), |
|||
], |
|||
), |
|||
migrations.CreateModel( |
|||
name="Question", |
|||
fields=[ |
|||
( |
|||
"id", |
|||
models.AutoField( |
|||
auto_created=True, |
|||
primary_key=True, |
|||
serialize=False, |
|||
verbose_name="ID", |
|||
), |
|||
), |
|||
("question_text", models.CharField(max_length=200)), |
|||
("pub_date", models.DateTimeField(verbose_name="date published")), |
|||
], |
|||
), |
|||
migrations.AddField( |
|||
model_name="choice", |
|||
name="question", |
|||
field=models.ForeignKey( |
|||
on_delete=django.db.models.deletion.CASCADE, to="polls.Question" |
|||
), |
|||
), |
|||
] |
@ -0,0 +1,12 @@ |
|||
from django.db import models |
|||
|
|||
|
|||
class Question(models.Model): |
|||
question_text = models.CharField(max_length=200) |
|||
pub_date = models.DateTimeField("date published") |
|||
|
|||
|
|||
class Choice(models.Model): |
|||
question = models.ForeignKey(Question, on_delete=models.CASCADE) |
|||
choice_text = models.CharField(max_length=200) |
|||
votes = models.IntegerField(default=0) |
@ -0,0 +1,37 @@ |
|||
import os |
|||
|
|||
import django |
|||
from asgiref.sync import sync_to_async |
|||
from fastapi import FastAPI |
|||
|
|||
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mysite.settings") |
|||
django.setup() |
|||
|
|||
from polls.models import Question # noqa: I001 E402 |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/questions") |
|||
def get_questions(): |
|||
questions = Question.objects.all() |
|||
|
|||
return [{"question": question.question_text} for question in questions] |
|||
|
|||
|
|||
@app.get("/questions-async") |
|||
async def get_questions_async(): |
|||
def _fetch_questions(): |
|||
return list(Question.objects.all()) |
|||
|
|||
questions = await sync_to_async(_fetch_questions)() |
|||
|
|||
return [{"question": question.question_text} for question in questions] |
|||
|
|||
|
|||
@app.get("/questions/{question_id}") |
|||
async def get_question(question_id: int): |
|||
question = await Question.objects.filter(id=question_id).afirst() |
|||
|
|||
return {"question": question.question_text} |
@ -1,4 +1,5 @@ |
|||
# For mkdocstrings and tests |
|||
httpx >=0.23.0,<0.28.0 |
|||
django |
|||
# For linting and generating docs versions |
|||
ruff ==0.11.2 |
|||
|
@ -0,0 +1,68 @@ |
|||
import pathlib |
|||
import sys |
|||
|
|||
import pytest |
|||
from django.core.management.color import no_style |
|||
from django.core.management.sql import sql_flush |
|||
from django.db import connection |
|||
from django.utils import timezone |
|||
from fastapi.testclient import TestClient |
|||
|
|||
HERE = pathlib.Path(__file__).parent |
|||
|
|||
sys.path.append(str(HERE.parents[2] / "docs_src" / "django_orm")) |
|||
|
|||
from docs_src.django_orm.tutorial001 import Question, app # noqa: I001 E402 |
|||
|
|||
|
|||
client = TestClient(app) |
|||
|
|||
|
|||
@pytest.fixture(scope="session", autouse=True) |
|||
def django_db_setup(): |
|||
connection.creation.create_test_db(verbosity=0, autoclobber=True) |
|||
|
|||
yield |
|||
|
|||
connection.creation.destroy_test_db("default", verbosity=0) |
|||
|
|||
|
|||
@pytest.fixture(autouse=True) |
|||
def flush_db(): |
|||
sql_list = sql_flush(no_style(), connection, allow_cascade=False) |
|||
|
|||
connection.ops.execute_sql_flush(sql_list) |
|||
|
|||
|
|||
def test_get_questions(): |
|||
Question.objects.create(question_text="there goes my hero", pub_date=timezone.now()) |
|||
|
|||
response = client.get("/questions") |
|||
|
|||
assert response.status_code == 200, response.text |
|||
assert response.json() == [{"question": "there goes my hero"}] |
|||
|
|||
|
|||
def test_question_empty(): |
|||
response = client.get("/questions") |
|||
|
|||
assert response.status_code == 200, response.text |
|||
assert response.json() == [] |
|||
|
|||
|
|||
def test_get_questions_async(): |
|||
Question.objects.create(question_text="everlong", pub_date=timezone.now()) |
|||
|
|||
response = client.get("/questions-async") |
|||
|
|||
assert response.status_code == 200, response.text |
|||
assert response.json() == [{"question": "everlong"}] |
|||
|
|||
|
|||
def test_get_single_question(): |
|||
question = Question.objects.create(question_text="my hero", pub_date=timezone.now()) |
|||
|
|||
response = client.get(f"/questions/{question.id}") |
|||
|
|||
assert response.status_code == 200, response.text |
|||
assert response.json() == {"question": "my hero"} |
Loading…
Reference in new issue