Browse Source
* ✨ Add docs and tests for encode/databases * ➕ Add testing-only dependency, databasespull/108/head
Sebastián Ramírez
6 years ago
committed by
GitHub
GitHub
9 changed files with 431 additions and 20 deletions
|
After Width: | Height: | Size: 69 KiB |
@ -0,0 +1,65 @@ |
|||
from typing import List |
|||
|
|||
import databases |
|||
import sqlalchemy |
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
|
|||
# SQLAlchemy specific code, as with any other app |
|||
DATABASE_URL = "sqlite:///./test.db" |
|||
# DATABASE_URL = "postgresql://user:password@postgresserver/db" |
|||
|
|||
database = databases.Database(DATABASE_URL) |
|||
|
|||
metadata = sqlalchemy.MetaData() |
|||
|
|||
notes = sqlalchemy.Table( |
|||
"notes", |
|||
metadata, |
|||
sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True), |
|||
sqlalchemy.Column("text", sqlalchemy.String), |
|||
sqlalchemy.Column("completed", sqlalchemy.Boolean), |
|||
) |
|||
|
|||
|
|||
engine = sqlalchemy.create_engine( |
|||
DATABASE_URL, connect_args={"check_same_thread": False} |
|||
) |
|||
metadata.create_all(engine) |
|||
|
|||
|
|||
class NoteIn(BaseModel): |
|||
text: str |
|||
completed: bool |
|||
|
|||
|
|||
class Note(BaseModel): |
|||
id: int |
|||
text: str |
|||
completed: bool |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.on_event("startup") |
|||
async def startup(): |
|||
await database.connect() |
|||
|
|||
|
|||
@app.on_event("shutdown") |
|||
async def shutdown(): |
|||
await database.disconnect() |
|||
|
|||
|
|||
@app.get("/notes/", response_model=List[Note]) |
|||
async def read_notes(): |
|||
query = notes.select() |
|||
return await database.fetch_all(query) |
|||
|
|||
|
|||
@app.post("/notes/", response_model=Note) |
|||
async def create_note(note: NoteIn): |
|||
query = notes.insert().values(text=note.text, completed=note.completed) |
|||
last_record_id = await database.execute(query) |
|||
return {**note.dict(), "id": last_record_id} |
|||
@ -0,0 +1,160 @@ |
|||
You can also use <a href="https://github.com/encode/databases" target="_blank">`encode/databases`</a> with **FastAPI** to connect to databases using `async` and `await`. |
|||
|
|||
It is compatible with: |
|||
|
|||
* PostgreSQL |
|||
* MySQL |
|||
* SQLite |
|||
|
|||
In this example, we'll use **SQLite**, because it uses a single file and Python has integrated support. So, you can copy this example and run it as is. |
|||
|
|||
Later, for your production application, you might want to use a database server like **PostgreSQL**. |
|||
|
|||
!!! tip |
|||
You could adopt ideas from the previous section about <a href="/tutorial/sql-databases/" target="_blank">SQLAlchemy ORM</a>, like using utility functions to perform operations in the database, independent of your **FastAPI** code. |
|||
|
|||
This section doesn't apply those ideas, to be equivalent to the counterpart in <a href="https://www.starlette.io/database/" target="_blank">Starlette</a>. |
|||
|
|||
## Import and set up `SQLAlchemy` |
|||
|
|||
* Import `SQLAlchemy`. |
|||
* Create a `metadata` object. |
|||
* Create a table `notes` using the `metadata` object. |
|||
|
|||
```Python hl_lines="4 14 16 17 18 19 20 21 22" |
|||
{!./src/async_sql_databases/tutorial001.py!} |
|||
``` |
|||
|
|||
!!! tip |
|||
Notice that all this code is pure SQLAlchemy Core. |
|||
|
|||
`databases` is not doing anything here yet. |
|||
|
|||
## Import and set up `databases` |
|||
|
|||
* Import `databases`. |
|||
* Create a `DATABASE_URL`. |
|||
* Create a `database` object. |
|||
|
|||
```Python hl_lines="3 9 12" |
|||
{!./src/async_sql_databases/tutorial001.py!} |
|||
``` |
|||
|
|||
!!! tip |
|||
If you where connecting to a different database (e.g. PostgreSQL), you would need to change the `DATABASE_URL`. |
|||
|
|||
## Create the tables |
|||
|
|||
In this case, we are creating the tables in the same Python file, but in production, you would probably want to create them with Alembic, integrated with migrations, etc. |
|||
|
|||
Here, this section would run directly, right before starting your **FastAPI** application. |
|||
|
|||
* Create an `engine`. |
|||
* Create all the tables from the `metadata` object. |
|||
|
|||
```Python hl_lines="25 26 27 28" |
|||
{!./src/async_sql_databases/tutorial001.py!} |
|||
``` |
|||
|
|||
## Create models |
|||
|
|||
Create Pydantic models for: |
|||
|
|||
* Notes to be created (`NoteIn`). |
|||
* Notes to be returned (`Note`). |
|||
|
|||
```Python hl_lines="31 32 33 36 37 38 39" |
|||
{!./src/async_sql_databases/tutorial001.py!} |
|||
``` |
|||
|
|||
By creating these Pydantic models, the input data will be validated, serialized (converted), and annotated (documented). |
|||
|
|||
So, you will be able to see it all in the interactive API docs. |
|||
|
|||
## Connect and disconnect |
|||
|
|||
* Create your `FastAPI` application. |
|||
* Create event handlers to connect and disconnect from the database. |
|||
|
|||
```Python hl_lines="42 45 46 47 50 51 52" |
|||
{!./src/async_sql_databases/tutorial001.py!} |
|||
``` |
|||
|
|||
## Read notes |
|||
|
|||
Create the *path operation function* to read notes: |
|||
|
|||
```Python hl_lines="55 56 57 58" |
|||
{!./src/async_sql_databases/tutorial001.py!} |
|||
``` |
|||
|
|||
!!! Note |
|||
Notice that as we communicate with the database using `await`, the *path operation function* is declared with `async`. |
|||
|
|||
### Notice the `response_model=List[Note]` |
|||
|
|||
It uses `typing.List`. |
|||
|
|||
That documents (and validates, serializes, filters) the output data, as a `list` of `Note`s. |
|||
|
|||
## Create notes |
|||
|
|||
Create the *path operation function* to create notes: |
|||
|
|||
```Python hl_lines="61 62 63 64 65" |
|||
{!./src/async_sql_databases/tutorial001.py!} |
|||
``` |
|||
|
|||
!!! Note |
|||
Notice that as we communicate with the database using `await`, the *path operation function* is declared with `async`. |
|||
|
|||
### About `{**note.dict(), "id": last_record_id}` |
|||
|
|||
`note` is a Pydantic `Note` object. |
|||
|
|||
`note.dict()` returns a `dict` with its data, something like: |
|||
|
|||
```Python |
|||
{ |
|||
"text": "Some note", |
|||
"completed": False, |
|||
} |
|||
``` |
|||
|
|||
but it doesn't have the `id` field. |
|||
|
|||
So we create a new `dict`, that contains the key-value pairs from `note.dict()` with: |
|||
|
|||
```Python |
|||
{**note.dict()} |
|||
``` |
|||
|
|||
`**note.dict()` "unpacks" the key value pairs directly, so, `{**note.dict()}` would be, more or less, a copy of `note.dict()`. |
|||
|
|||
And then, we extend that copy `dict`, adding another key-value pair: `"id": last_record_id`: |
|||
|
|||
```Python |
|||
{**note.dict(), "id": last_record_id} |
|||
``` |
|||
|
|||
So, the final result returned would be something like: |
|||
|
|||
```Python |
|||
{ |
|||
"id": 1, |
|||
"text": "Some note", |
|||
"completed": False, |
|||
} |
|||
``` |
|||
|
|||
## Check it |
|||
|
|||
You can copy this code as is, and see the docs at <a href="http://127.0.0.1:8000/docs" target="_blank">http://127.0.0.1:8000/docs</a>. |
|||
|
|||
There you can see all your API documented and interact with it: |
|||
|
|||
<img src="/img/tutorial/async-sql-databases/image01.png"> |
|||
|
|||
## More info |
|||
|
|||
You can read more about <a href="https://github.com/encode/databases" target="_blank">`encode/databases` at its GitHub page</a>. |
|||
@ -0,0 +1,131 @@ |
|||
from starlette.testclient import TestClient |
|||
|
|||
from async_sql_databases.tutorial001 import app |
|||
|
|||
openapi_schema = { |
|||
"openapi": "3.0.2", |
|||
"info": {"title": "Fast API", "version": "0.1.0"}, |
|||
"paths": { |
|||
"/notes/": { |
|||
"get": { |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"title": "Response_Read_Notes", |
|||
"type": "array", |
|||
"items": {"$ref": "#/components/schemas/Note"}, |
|||
} |
|||
} |
|||
}, |
|||
} |
|||
}, |
|||
"summary": "Read Notes Get", |
|||
"operationId": "read_notes_notes__get", |
|||
}, |
|||
"post": { |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": {"$ref": "#/components/schemas/Note"} |
|||
} |
|||
}, |
|||
}, |
|||
"422": { |
|||
"description": "Validation Error", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/HTTPValidationError" |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
}, |
|||
"summary": "Create Note Post", |
|||
"operationId": "create_note_notes__post", |
|||
"requestBody": { |
|||
"content": { |
|||
"application/json": { |
|||
"schema": {"$ref": "#/components/schemas/NoteIn"} |
|||
} |
|||
}, |
|||
"required": True, |
|||
}, |
|||
}, |
|||
} |
|||
}, |
|||
"components": { |
|||
"schemas": { |
|||
"NoteIn": { |
|||
"title": "NoteIn", |
|||
"required": ["text", "completed"], |
|||
"type": "object", |
|||
"properties": { |
|||
"text": {"title": "Text", "type": "string"}, |
|||
"completed": {"title": "Completed", "type": "boolean"}, |
|||
}, |
|||
}, |
|||
"Note": { |
|||
"title": "Note", |
|||
"required": ["id", "text", "completed"], |
|||
"type": "object", |
|||
"properties": { |
|||
"id": {"title": "Id", "type": "integer"}, |
|||
"text": {"title": "Text", "type": "string"}, |
|||
"completed": {"title": "Completed", "type": "boolean"}, |
|||
}, |
|||
}, |
|||
"ValidationError": { |
|||
"title": "ValidationError", |
|||
"required": ["loc", "msg", "type"], |
|||
"type": "object", |
|||
"properties": { |
|||
"loc": { |
|||
"title": "Location", |
|||
"type": "array", |
|||
"items": {"type": "string"}, |
|||
}, |
|||
"msg": {"title": "Message", "type": "string"}, |
|||
"type": {"title": "Error Type", "type": "string"}, |
|||
}, |
|||
}, |
|||
"HTTPValidationError": { |
|||
"title": "HTTPValidationError", |
|||
"type": "object", |
|||
"properties": { |
|||
"detail": { |
|||
"title": "Detail", |
|||
"type": "array", |
|||
"items": {"$ref": "#/components/schemas/ValidationError"}, |
|||
} |
|||
}, |
|||
}, |
|||
} |
|||
}, |
|||
} |
|||
|
|||
|
|||
def test_openapi_schema(): |
|||
with TestClient(app) as client: |
|||
response = client.get("/openapi.json") |
|||
assert response.status_code == 200 |
|||
assert response.json() == openapi_schema |
|||
|
|||
|
|||
def test_create_read(): |
|||
with TestClient(app) as client: |
|||
note = {"text": "Foo bar", "completed": False} |
|||
response = client.post("/notes/", json=note) |
|||
assert response.status_code == 200 |
|||
data = response.json() |
|||
assert data["text"] == note["text"] |
|||
assert data["completed"] == note["completed"] |
|||
assert "id" in data |
|||
response = client.get(f"/notes/") |
|||
assert response.status_code == 200 |
|||
assert data in response.json() |
|||
Loading…
Reference in new issue