Browse Source
* ✨ Implement support for Pydantic's ORM mode * 🏗️ Re-structure/augment SQL tutorial source using ORM mode * 📝 Update SQL docs with SQLAlchemy, ORM mode, relationships * 🔥 Remove unused util in tutorial * 📝 Add tutorials for simple dict bodies and responses * 🔥 Remove old SQL tutorial * ✅ Add/update tests for SQL tutorial * ✅ Add tests for simple dicts (body and response) * 🐛 Fix cloning field from original fieldpull/330/head
Sebastián Ramírez
6 years ago
committed by
GitHub
GitHub
19 changed files with 1141 additions and 226 deletions
@ -0,0 +1,10 @@ |
|||
from typing import Dict |
|||
|
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.post("/index-weights/") |
|||
async def create_index_weights(weights: Dict[int, float]): |
|||
return weights |
|||
@ -0,0 +1,10 @@ |
|||
from typing import Dict |
|||
|
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/keyword-weights/", response_model=Dict[str, float]) |
|||
async def read_keyword_weights(): |
|||
return {"foo": 2.3, "bar": 3.4} |
|||
@ -0,0 +1,36 @@ |
|||
from sqlalchemy.orm import Session |
|||
|
|||
from . import models, schemas |
|||
|
|||
|
|||
def get_user(db: Session, user_id: int): |
|||
return db.query(models.User).filter(models.User.id == user_id).first() |
|||
|
|||
|
|||
def get_user_by_email(db: Session, email: str): |
|||
return db.query(models.User).filter(models.User.email == email).first() |
|||
|
|||
|
|||
def get_users(db: Session, skip: int = 0, limit: int = 100): |
|||
return db.query(models.User).offset(skip).limit(limit).all() |
|||
|
|||
|
|||
def create_user(db: Session, user: schemas.UserCreate): |
|||
fake_hashed_password = user.password + "notreallyhashed" |
|||
db_user = models.User(email=user.email, hashed_password=fake_hashed_password) |
|||
db.add(db_user) |
|||
db.commit() |
|||
db.refresh(db_user) |
|||
return db_user |
|||
|
|||
|
|||
def get_items(db: Session, skip: int = 0, limit: int = 100): |
|||
return db.query(models.Item).offset(skip).limit(limit).all() |
|||
|
|||
|
|||
def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int): |
|||
db_item = models.Item(**item.dict(), owner_id=user_id) |
|||
db.add(db_item) |
|||
db.commit() |
|||
db.refresh(db_item) |
|||
return db_item |
|||
@ -0,0 +1,13 @@ |
|||
from sqlalchemy import create_engine |
|||
from sqlalchemy.ext.declarative import declarative_base |
|||
from sqlalchemy.orm import sessionmaker |
|||
|
|||
SQLALCHEMY_DATABASE_URI = "sqlite:///./test.db" |
|||
# SQLALCHEMY_DATABASE_URI = "postgresql://user:password@postgresserver/db" |
|||
|
|||
engine = create_engine( |
|||
SQLALCHEMY_DATABASE_URI, connect_args={"check_same_thread": False} |
|||
) |
|||
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) |
|||
|
|||
Base = declarative_base() |
|||
@ -0,0 +1,64 @@ |
|||
from typing import List |
|||
|
|||
from fastapi import Depends, FastAPI, HTTPException |
|||
from sqlalchemy.orm import Session |
|||
from starlette.requests import Request |
|||
from starlette.responses import Response |
|||
|
|||
from . import crud, models, schemas |
|||
from .database import SessionLocal, engine |
|||
|
|||
models.Base.metadata.create_all(bind=engine) |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.middleware("http") |
|||
async def db_session_middleware(request: Request, call_next): |
|||
response = Response("Internal server error", status_code=500) |
|||
try: |
|||
request.state.db = SessionLocal() |
|||
response = await call_next(request) |
|||
finally: |
|||
request.state.db.close() |
|||
return response |
|||
|
|||
|
|||
# Dependency |
|||
def get_db(request: Request): |
|||
return request.state.db |
|||
|
|||
|
|||
@app.post("/users/", response_model=schemas.User) |
|||
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)): |
|||
db_user = crud.get_user_by_email(db, email=user.email) |
|||
if db_user: |
|||
raise HTTPException(status_code=400, detail="Email already registered") |
|||
return crud.create_user(db=db, user=user) |
|||
|
|||
|
|||
@app.get("/users/", response_model=List[schemas.User]) |
|||
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): |
|||
users = crud.get_users(db, skip=skip, limit=limit) |
|||
return users |
|||
|
|||
|
|||
@app.get("/users/{user_id}", response_model=schemas.User) |
|||
def read_user(user_id: int, db: Session = Depends(get_db)): |
|||
db_user = crud.get_user(db, user_id=user_id) |
|||
if db_user is None: |
|||
raise HTTPException(status_code=404, detail="User not found") |
|||
return db_user |
|||
|
|||
|
|||
@app.post("/users/{user_id}/items/", response_model=schemas.Item) |
|||
def create_item_for_user( |
|||
user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db) |
|||
): |
|||
return crud.create_user_item(db=db, item=item, user_id=user_id) |
|||
|
|||
|
|||
@app.get("/items/", response_model=List[schemas.Item]) |
|||
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): |
|||
items = crud.get_items(db, skip=skip, limit=limit) |
|||
return items |
|||
@ -0,0 +1,26 @@ |
|||
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String |
|||
from sqlalchemy.orm import relationship |
|||
|
|||
from .database import Base |
|||
|
|||
|
|||
class User(Base): |
|||
__tablename__ = "users" |
|||
|
|||
id = Column(Integer, primary_key=True, index=True) |
|||
email = Column(String, unique=True, index=True) |
|||
hashed_password = Column(String) |
|||
is_active = Column(Boolean, default=True) |
|||
|
|||
items = relationship("Item", back_populates="owner") |
|||
|
|||
|
|||
class Item(Base): |
|||
__tablename__ = "items" |
|||
|
|||
id = Column(Integer, primary_key=True, index=True) |
|||
title = Column(String, index=True) |
|||
description = Column(String, index=True) |
|||
owner_id = Column(Integer, ForeignKey("users.id")) |
|||
|
|||
owner = relationship("User", back_populates="items") |
|||
@ -0,0 +1,37 @@ |
|||
from typing import List |
|||
|
|||
from pydantic import BaseModel |
|||
|
|||
|
|||
class ItemBase(BaseModel): |
|||
title: str |
|||
description: str = None |
|||
|
|||
|
|||
class ItemCreate(ItemBase): |
|||
pass |
|||
|
|||
|
|||
class Item(ItemBase): |
|||
id: int |
|||
owner_id: int |
|||
|
|||
class Config: |
|||
orm_mode = True |
|||
|
|||
|
|||
class UserBase(BaseModel): |
|||
email: str |
|||
|
|||
|
|||
class UserCreate(UserBase): |
|||
password: str |
|||
|
|||
|
|||
class User(UserBase): |
|||
id: int |
|||
is_active: bool |
|||
items: List[Item] = [] |
|||
|
|||
class Config: |
|||
orm_mode = True |
|||
@ -1,76 +0,0 @@ |
|||
from fastapi import Depends, FastAPI |
|||
from sqlalchemy import Boolean, Column, Integer, String, create_engine |
|||
from sqlalchemy.ext.declarative import declarative_base, declared_attr |
|||
from sqlalchemy.orm import Session, sessionmaker |
|||
from starlette.requests import Request |
|||
from starlette.responses import Response |
|||
|
|||
# SQLAlchemy specific code, as with any other app |
|||
SQLALCHEMY_DATABASE_URI = "sqlite:///./test.db" |
|||
# SQLALCHEMY_DATABASE_URI = "postgresql://user:password@postgresserver/db" |
|||
|
|||
engine = create_engine( |
|||
SQLALCHEMY_DATABASE_URI, connect_args={"check_same_thread": False} |
|||
) |
|||
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) |
|||
|
|||
|
|||
class CustomBase: |
|||
# Generate __tablename__ automatically |
|||
@declared_attr |
|||
def __tablename__(cls): |
|||
return cls.__name__.lower() |
|||
|
|||
|
|||
Base = declarative_base(cls=CustomBase) |
|||
|
|||
|
|||
class User(Base): |
|||
id = Column(Integer, primary_key=True, index=True) |
|||
email = Column(String, unique=True, index=True) |
|||
hashed_password = Column(String) |
|||
is_active = Column(Boolean(), default=True) |
|||
|
|||
|
|||
Base.metadata.create_all(bind=engine) |
|||
|
|||
db_session = SessionLocal() |
|||
|
|||
first_user = db_session.query(User).first() |
|||
if not first_user: |
|||
u = User(email="[email protected]", hashed_password="notreallyhashed") |
|||
db_session.add(u) |
|||
db_session.commit() |
|||
|
|||
db_session.close() |
|||
|
|||
|
|||
# Utility |
|||
def get_user(db_session: Session, user_id: int): |
|||
return db_session.query(User).filter(User.id == user_id).first() |
|||
|
|||
|
|||
# Dependency |
|||
def get_db(request: Request): |
|||
return request.state.db |
|||
|
|||
|
|||
# FastAPI specific code |
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/users/{user_id}") |
|||
def read_user(user_id: int, db: Session = Depends(get_db)): |
|||
user = get_user(db, user_id=user_id) |
|||
return user |
|||
|
|||
|
|||
@app.middleware("http") |
|||
async def db_session_middleware(request: Request, call_next): |
|||
response = Response("Internal server error", status_code=500) |
|||
try: |
|||
request.state.db = SessionLocal() |
|||
response = await call_next(request) |
|||
finally: |
|||
request.state.db.close() |
|||
return response |
|||
@ -25,25 +25,77 @@ Later, for your production application, you might want to use a database server |
|||
|
|||
The **FastAPI** specific code is as small as always. |
|||
|
|||
## Import SQLAlchemy components |
|||
## ORMs |
|||
|
|||
For now, don't pay attention to the rest, only the imports: |
|||
**FastAPI** works with any database and any style of library to talk to the database. |
|||
|
|||
```Python hl_lines="2 3 4" |
|||
{!./src/sql_databases/tutorial001.py!} |
|||
A common pattern is to use an "ORM": an "object-relational mapping" library. |
|||
|
|||
An ORM has tools to convert ("*map*") between *objects* in code and database tables ("*relations*"). |
|||
|
|||
With an ORM, you normally create a class that represents a table in a SQL database, each attribute of the class represents a column, with a name and a type. |
|||
|
|||
For example a class `Pet` could represent a SQL table `pets`. |
|||
|
|||
And each *instance* object of that class represents a row in the database. |
|||
|
|||
For example an object `mr_furry` (an instance of `Pet`) could have an attribute `mr_furry.type`, for the column `type`. And the value of that attribute could be, e.g. `"cat"`. |
|||
|
|||
These ORMs also have tools to make the connections or relations between tables or entities. |
|||
|
|||
This way, you could also have an attribute `mr_furry.owner` and the owner would contain the data for this pet's owner, taken from the table *owners*. |
|||
|
|||
So, `mr_furry.owner.name` could be the name (from the `name` column in the `owners` table) of this pet's owner. |
|||
|
|||
It could have a value like `"Alice"`. |
|||
|
|||
And the ORM will do all the work to get the information from the corresponding table *owners* when you try to access it from your pet object. |
|||
|
|||
Common ORMs are for example: Django-ORM (part of the Django framework), SQLAlchemy ORM (part of SQLAlchemy, independent of framework) and Peewee (independent of framework), among others. |
|||
|
|||
Here we will see how to work with **SQLAlchemy ORM**. |
|||
|
|||
The same way, you could use Peewee or any other. |
|||
|
|||
## File structure |
|||
|
|||
For these examples, let's say you have a directory `sql_app` with a structure like this: |
|||
|
|||
``` |
|||
├── sql_app |
|||
│ ├── __init__.py |
|||
│ ├── crud.py |
|||
│ ├── database.py |
|||
│ ├── main.py |
|||
│ ├── models.py |
|||
│ ├── schemas.py |
|||
``` |
|||
|
|||
## Define the database |
|||
The file `__init__.py` is just an empty file, but it tells Python that `sql_app` with all its modules (Python files) is a package. |
|||
|
|||
Now let's see what each file/module does. |
|||
|
|||
## Create the SQLAlchemy parts |
|||
|
|||
Let's see the file `sql_app/database.py`. |
|||
|
|||
Define the database that SQLAlchemy should "connect" to: |
|||
### Import the SQLAlchemy parts |
|||
|
|||
```Python hl_lines="9" |
|||
{!./src/sql_databases/tutorial001.py!} |
|||
```Python hl_lines="1 2 3" |
|||
{!./src/sql_databases/sql_app/database.py!} |
|||
``` |
|||
|
|||
### Create a database URL for SQLAlchemy |
|||
|
|||
```Python hl_lines="5 6" |
|||
{!./src/sql_databases/sql_app/database.py!} |
|||
``` |
|||
|
|||
In this example, we are "connecting" to a SQLite database (opening a file with the SQLite database). |
|||
|
|||
The file will be located at the same directory in the file `test.db`. That's why the last part is `./test.db`. |
|||
The file will be located at the same directory in the file `test.db`. |
|||
|
|||
That's why the last part is `./test.db`. |
|||
|
|||
If you were using a **PostgreSQL** database instead, you would just have to uncomment the line: |
|||
|
|||
@ -54,16 +106,20 @@ SQLALCHEMY_DATABASE_URI = "postgresql://user:password@postgresserver/db" |
|||
...and adapt it with your database data and credentials (equivalently for MySQL, MariaDB or any other). |
|||
|
|||
!!! tip |
|||
|
|||
|
|||
This is the main line that you would have to modify if you wanted to use a different database. |
|||
|
|||
## Create the SQLAlchemy `engine` |
|||
### Create the SQLAlchemy `engine` |
|||
|
|||
```Python hl_lines="12 13 14" |
|||
{!./src/sql_databases/tutorial001.py!} |
|||
The first step is to create a SQLAlchemy "engine". |
|||
|
|||
We will later use this `engine` in other places. |
|||
|
|||
```Python hl_lines="8 9 10" |
|||
{!./src/sql_databases/sql_app/database.py!} |
|||
``` |
|||
|
|||
### Note |
|||
#### Note |
|||
|
|||
The argument: |
|||
|
|||
@ -78,144 +134,353 @@ connect_args={"check_same_thread": False} |
|||
That argument `check_same_thread` is there mainly to be able to run the tests that cover this example. |
|||
|
|||
|
|||
## Create a `SessionLocal` class |
|||
### Create a `SessionLocal` class |
|||
|
|||
Each instance of the `SessionLocal` class will have a connection to the database. |
|||
Each instance of the `SessionLocal` class will have a session/connection to the database. |
|||
|
|||
This object (class) is not a connection to the database yet, but once we create an instance of this class, that instance will have the actual connection to the database. |
|||
This object (class) is not a session/connection to the database yet, but once we create an instance of this class, that instance will have the actual connection to the database. |
|||
|
|||
We name it `SessionLocal` to distinguish it from the `Session` we are importing from SQLAlchemy. |
|||
|
|||
We will use `Session` to declare types later and to get better editor support and completion. |
|||
We will use `Session` (the one imported from SQLAlchemy) later. |
|||
|
|||
For now, create the `SessionLocal`: |
|||
To create the `SessionLocal` class, use the function `sessionmaker`: |
|||
|
|||
```Python hl_lines="15" |
|||
{!./src/sql_databases/tutorial001.py!} |
|||
```Python hl_lines="11" |
|||
{!./src/sql_databases/sql_app/database.py!} |
|||
``` |
|||
|
|||
## Create a middleware to handle sessions |
|||
### Create a `Base` class |
|||
|
|||
Now let's temporarily jump to the end of the file, to use the `SessionLocal` class we created above. |
|||
Now use the function `declarative_base()` that returns a class. |
|||
|
|||
We need to have an independent database session/connection (`SessionLocal`) per request, use the same session through all the request and then close it after the request is finished. |
|||
Later we will inherit from this class to create each of the database models or classes (the ORM models): |
|||
|
|||
And then a new session will be created for the next request. |
|||
```Python hl_lines="13" |
|||
{!./src/sql_databases/sql_app/database.py!} |
|||
``` |
|||
|
|||
For that, we will create a new middleware. |
|||
## Create the database models |
|||
|
|||
A "middleware" is a function that is always executed for each request, and have code before and after the request. |
|||
Let's now see the file `sql_app/models.py`. |
|||
|
|||
This middleware (just a function) will create a new SQLAlchemy `SessionLocal` for each request, add it to the request and then close it once the request is finished. |
|||
### Create SQLAlchemy models from the `Base` class |
|||
|
|||
We will use this `Base` class we created before to create the SQLAlchemy models. |
|||
|
|||
!!! tip |
|||
SQLAlchemy uses the term "**model**" to refer to these classes and instances that interact with the database. |
|||
|
|||
But Pydantic also uses the term "**model**" to refer to something different, the data validation, conversion, and documentation classes and instances. |
|||
|
|||
Import `Base` from `database` (the file `database.py` from above). |
|||
|
|||
```Python hl_lines="68 69 70 71 72 73 74 75 76" |
|||
{!./src/sql_databases/tutorial001.py!} |
|||
Create classes that inherit from it. |
|||
|
|||
These classes are the SQLAlchemy models. |
|||
|
|||
```Python hl_lines="4 7 8 18 19" |
|||
{!./src/sql_databases/sql_app/models.py!} |
|||
``` |
|||
|
|||
!!! info |
|||
We put the creation of the `SessionLocal()` and handling of the requests in a `try` block. |
|||
The `__tablename__` attribute tells SQLAlchemy the name of the table to use in the database for each of these models. |
|||
|
|||
And then we close it in the `finally` block. |
|||
|
|||
This way we make sure the database session is always closed after the request. Even if there was an exception in the middle. |
|||
### Create model attributes/columns |
|||
|
|||
### About `request.state` |
|||
Now create all the model (class) attributes. |
|||
|
|||
<a href="https://www.starlette.io/requests/#other-state" target="_blank">`request.state` is a property of each Starlette `Request` object</a>, it is there to store arbitrary objects attached to the request itself, like the database session in this case. |
|||
Each of these attributes represents a column in its corresponding database table. |
|||
|
|||
For us in this case, it helps us ensuring a single session/database-connection is used through all the request, and then closed afterwards (in the middleware). |
|||
We use `Column` from SQLAlchemy as the default value. |
|||
|
|||
## Create a dependency |
|||
And we pass a SQLAlchemy class "type", as `Integer`, `String`, and `Boolean`, that defines the type in the database, as an argument. |
|||
|
|||
To simplify the code, reduce repetition and get better editor support, we will create a dependency that returns this same database session from the request. |
|||
```Python hl_lines="1 10 11 12 13 21 22 23 24" |
|||
{!./src/sql_databases/sql_app/models.py!} |
|||
``` |
|||
|
|||
And when using the dependency in a path operation function, we declare it with the type `Session` we imported directly from SQLAlchemy. |
|||
### Create the relationships |
|||
|
|||
This will then give us better editor support inside the path operation function, because the editor will know that the `db` parameter is of type `Session`. |
|||
Now create the relationships. |
|||
|
|||
For this, we use `relationship` provided by SQLAlchemy ORM. |
|||
|
|||
```Python hl_lines="54 55 69" |
|||
{!./src/sql_databases/tutorial001.py!} |
|||
This will become, more or less, a "magic" attribute that will contain the values from other tables related to this one. |
|||
|
|||
```Python hl_lines="2 15 26" |
|||
{!./src/sql_databases/sql_app/models.py!} |
|||
``` |
|||
|
|||
!!! info "Technical Details" |
|||
The parameter `db` is actually of type `SessionLocal`, but this class (created with `sessionmaker()`) is a "proxy" of a SQLAlchemy `Session`, so, the editor doesn't really know what methods are provided. |
|||
When accessing the attribute `items` in a `User`, as in `my_user.items`, it will have a list of `Item` SQLAlchemy models (from the `items` table) that have a foreign key pointing to this record in the `users` table. |
|||
|
|||
When you access `my_user.items`, SQLAlchemy will actually go and fetch the items from the database in the `items` table and populate them here. |
|||
|
|||
And when accessing the attribute `owner` in an `Item`, it will contain a `User` SQLAlchemy model from the `users` table. It will use the `owner_id` attribute/column with its foreign key to know which record to get from the `users` table. |
|||
|
|||
## Create the Pydantic models |
|||
|
|||
Now let's check the file `sql_app/schemas.py`. |
|||
|
|||
!!! tip |
|||
To avoid confusion between the SQLAlchemy *models* and the Pydantic *models*, we will have the file `models.py` with the SQLAlchemy models, and the file `schemas.py` with the Pydantic models. |
|||
|
|||
These Pydantic models define more or less a "schema" (a valid data shape). |
|||
|
|||
But by declaring the type as `Session`, the editor now can know the available methods (`.add()`, `.query()`, `.commit()`, etc) and can provide better support (like completion). The type declaration doesn't affect the actual object. |
|||
So this will help us avoiding confusion while using both. |
|||
|
|||
### Create initial Pydantic *models* / schemas |
|||
|
|||
Create an `ItemBase` and `UserBase` Pydantic *models* (or let's say "schemas") to have common attributes while creating or reading data. |
|||
|
|||
## Create a `CustomBase` model |
|||
And create an `ItemCreate` and `UserCreate` that inherit from them (so they will have the same attributes), plus any additional data (attributes) needed for creation. |
|||
|
|||
This is more of a trick to facilitate your life than something required. |
|||
So, the user will also have a `password` when creating it. |
|||
|
|||
But by creating this `CustomBase` class and inheriting from it, your models will have automatic `__tablename__` attributes (that are required by SQLAlchemy). |
|||
But for security, the `password` won't be in other Pydantic *models*, for example, it won't be sent from the API when reading a user. |
|||
|
|||
That way you don't have to declare them explicitly in every model. |
|||
```Python hl_lines="3 6 7 8 11 12 23 24 27 28" |
|||
{!./src/sql_databases/sql_app/schemas.py!} |
|||
``` |
|||
|
|||
#### SQLAlchemy style and Pydantic style |
|||
|
|||
So, your models will behave very similarly to, for example, Flask-SQLAlchemy. |
|||
Notice that SQLAlchemy *models* define attributes using `=`, and pass the type as a parameter to `Column`, like in: |
|||
|
|||
```Python hl_lines="18 19 20 21 22" |
|||
{!./src/sql_databases/tutorial001.py!} |
|||
```Python |
|||
name = Column(String) |
|||
``` |
|||
|
|||
## Create the SQLAlchemy `Base` model |
|||
while Pydantic *models* declare the types using `:`, the new type annotation syntax/type hints: |
|||
|
|||
```Python hl_lines="25" |
|||
{!./src/sql_databases/tutorial001.py!} |
|||
```Python |
|||
name: str |
|||
``` |
|||
|
|||
## Create your application data model |
|||
Have it in mind, so you don't get confused when using `=` and `:` with them. |
|||
|
|||
### Create Pydantic *models* / schemas for reading / returning |
|||
|
|||
Now create Pydantic *models* (schemas) that will be used when reading data, when returning it from the API. |
|||
|
|||
For example, before creating an item, we don't know what will be the ID assigned to it, but when reading it (when returning it from the API) we will already know its ID. |
|||
|
|||
Now this is finally code specific to your app. |
|||
The same way, when reading a user, we can now declare that `items` will contain the items that belong to this user. |
|||
|
|||
Here's a user model that will be a table in the database: |
|||
Not only the IDs of those items, but all the data that we defined in the Pydantic *model* for reading items: `Item`. |
|||
|
|||
```Python hl_lines="28 29 30 31 32" |
|||
{!./src/sql_databases/tutorial001.py!} |
|||
```Python hl_lines="15 16 17 31 32 33 34" |
|||
{!./src/sql_databases/sql_app/schemas.py!} |
|||
``` |
|||
|
|||
## Initialize your application |
|||
!!! tip |
|||
Notice that the `User`, the Pydantic *model* that will be used when reading a user (returning it from the API) doesn't include the `password`. |
|||
|
|||
### Use Pydantic's `orm_mode` |
|||
|
|||
Now, in the Pydantic *models* for reading, `Item` and `User`, add an internal `Config` class. |
|||
|
|||
In a very simplistic way, initialize your database (create the tables, etc) and make sure you have a first user: |
|||
This <a href="https://pydantic-docs.helpmanual.io/#config" target="_blank">`Config`</a> class is used to provide configurations to Pydantic. |
|||
|
|||
```Python hl_lines="35 37 39 40 41 42 43 45" |
|||
{!./src/sql_databases/tutorial001.py!} |
|||
In the `Config` class, set the attribute `orm_mode = True`. |
|||
|
|||
```Python hl_lines="15 19 20 31 36 37" |
|||
{!./src/sql_databases/sql_app/schemas.py!} |
|||
``` |
|||
|
|||
!!! info |
|||
Notice that we close the session with `db_session.close()`. |
|||
!!! tip |
|||
Notice it's assigning a value with `=`, like: |
|||
|
|||
`orm_mode = True` |
|||
|
|||
It doesn't use `:` as for the type declarations before. |
|||
|
|||
This is setting a config value, not declaring a type. |
|||
|
|||
Pydantic's `orm_mode` will tell the Pydantic *model* to read the data even if it is not a `dict`, but an ORM model (or any other arbitrary object with attributes). |
|||
|
|||
This way, Instead of only trying to get the `id` value from a `dict`, as in: |
|||
|
|||
```Python |
|||
id = data["id"] |
|||
``` |
|||
|
|||
it will also try to get it from an attribute, as in: |
|||
|
|||
```Python |
|||
id = data.id |
|||
``` |
|||
|
|||
And with this, the Pydantic *model* is compatible with ORMs, and you can just declare it in the `response_model` argument in your *path operations*. |
|||
|
|||
You will be able to return a database model and it will read the data from it. |
|||
|
|||
#### Technical Details about ORM mode |
|||
|
|||
SQLAlchemy and many others are by default "lazy loading". |
|||
|
|||
That means, for example, that they don't fetch the data for relationships from the database unless you try to access the attribute that would contain that data. |
|||
|
|||
For example, accessing the attribute `items`: |
|||
|
|||
```Python |
|||
current_user.items |
|||
``` |
|||
|
|||
would make SQLAlchemy go to the `items` table and get the items for this user, but not before. |
|||
|
|||
Without `orm_mode`, if you returned a SQLAlchemy model from your *path operation*, it wouldn't include the relationship data. |
|||
|
|||
Even if you declared those relationships in your Pydantic models. |
|||
|
|||
But with ORM mode, as Pydantic itself will try to access the data it needs from attributes (instead of assuming a `dict`), you can declare the specific data you want to return and it will be able to go and get it, even from ORMs. |
|||
|
|||
We close this session because we only used it to create this first user. |
|||
## CRUD utils |
|||
|
|||
Every new request will get its own new session. |
|||
Now let's see the file `sql_app/crud.py`. |
|||
|
|||
### Note |
|||
In this file we will have reusable functions to interact with the data in the database. |
|||
|
|||
**CRUD** comes from: **C**reate, **R**ead, **U**pdate, and **D**elete. |
|||
|
|||
...although in this example we are only creating and reading. |
|||
|
|||
### Read data |
|||
|
|||
Import `Session` from `sqlalchemy.orm`, this will allow you to declare the type of the `db` parameters and have better type checks and completion in your functions. |
|||
|
|||
Import `models` (the SQLAlchemy models) and `schemas` (the Pydantic *models* / schemas). |
|||
|
|||
Create utility functions to: |
|||
|
|||
* Read a single user by ID and by email. |
|||
* Read multiple users. |
|||
* Read a single item. |
|||
|
|||
```Python hl_lines="1 3 6 7 10 11 14 15 27 28" |
|||
{!./src/sql_databases/sql_app/crud.py!} |
|||
``` |
|||
|
|||
!!! tip |
|||
By creating functions that are only dedicated to interacting with the database (get a user or an item) independent of your path operation function, you can more easily reuse them in multiple parts and also add <abbr title="Automated tests, written in code, that check if another piece of code is working correctly.">unit tests</abbr> for them. |
|||
|
|||
### Create data |
|||
|
|||
Now create utility functions to create data. |
|||
|
|||
The steps are: |
|||
|
|||
* Create a SQLAlchemy model *instance* with your data. |
|||
* `add` that instance object to your database session. |
|||
* `commit` the changes to the database (so that they are saved). |
|||
* `refresh` your instance (so that it contains any new data from the database, like the generated ID). |
|||
|
|||
```Python hl_lines="18 19 20 21 22 23 24 31 32 33 34 35 36" |
|||
{!./src/sql_databases/sql_app/crud.py!} |
|||
``` |
|||
|
|||
!!! tip |
|||
The SQLAlchemy model for `User` contains a `hashed_password` that should contain a secure hashed version of the password. |
|||
|
|||
But as what the API client provides is the original password, you need to extract it and generate the hashed password in your application. |
|||
|
|||
And then pass the `hashed_password` argument with the value to save. |
|||
|
|||
!!! warning |
|||
This example is not secure, the password is not hashed. |
|||
|
|||
In a real life application you would need to hash the password and never save them in plaintext. |
|||
|
|||
For more details, go back to the Security section in the tutorial. |
|||
|
|||
Here we are focusing only on the tools and mechanics of databases. |
|||
|
|||
!!! tip |
|||
Instead of passing each of the keyword arguments to `Item` and reading each one of them from the Pydantic *model*, we are generating a `dict` with the Pydantic *model*'s data with: |
|||
|
|||
`item.dict()` |
|||
|
|||
and then we are passing the `dict`'s key-value pairs as the keyword arguments to the SQLAlchemy `Item`, with: |
|||
|
|||
`Item(**item.dict())` |
|||
|
|||
And then we pass the extra keyword argument `owner_id` that is not provided by the Pydantic *model*, with: |
|||
|
|||
`Item(**item.dict(), owner_id=user_id)` |
|||
|
|||
## Main **FastAPI** app |
|||
|
|||
And now in the file `sql_app/main.py` let's integrate and use all the other parts we created before. |
|||
|
|||
### Create the database tables |
|||
|
|||
In a very simplistic way, create the database tables: |
|||
|
|||
```Python hl_lines="11" |
|||
{!./src/sql_databases/sql_app/main.py!} |
|||
``` |
|||
|
|||
#### Alembic Note |
|||
|
|||
Normally you would probably initialize your database (create tables, etc) with <a href="https://alembic.sqlalchemy.org/en/latest/" target="_blank">Alembic</a>. |
|||
|
|||
And you would also use Alembic for migrations (that's its main job). For whenever you change the structure of your database, add a new column, a new table, etc. |
|||
And you would also use Alembic for "migrations" (that's its main job). |
|||
|
|||
The same way, you would probably make sure there's a first user in an external script that runs before your application, or as part of the application startup. |
|||
A "migration" is the set of steps needed whenever you change the structure of your SQLAlchemy models, add a new attribute, etc. to replicate those changes in the database, add a new column, a new table, etc. |
|||
|
|||
In this example we are doing those two operations in a very simple way, directly in the code, to focus on the main points. |
|||
### Create a middleware to handle sessions |
|||
|
|||
Also, as all the functionality is self-contained in the same code, you can copy it and run it directly, and it will work as is. |
|||
Now use the `SessionLocal` class we created in the `sql_app/databases.py` file. |
|||
|
|||
We need to have an independent database session/connection (`SessionLocal`) per request, use the same session through all the request and then close it after the request is finished. |
|||
|
|||
And then a new session will be created for the next request. |
|||
|
|||
## Get a user |
|||
For that, we will create a new middleware. |
|||
|
|||
By creating a function that is only dedicated to getting your user from a `user_id` (or any other parameter) independent of your path operation function, you can more easily re-use it in multiple parts and also add <abbr title="Automated tests, written in code, that check if another piece of code is working correctly.">unit tests</abbr> for it: |
|||
A "middleware" is a function that is always executed for each request, and have code before and after the request. |
|||
|
|||
This middleware (just a function) will create a new SQLAlchemy `SessionLocal` for each request, add it to the request and then close it once the request is finished. |
|||
|
|||
```Python hl_lines="49 50" |
|||
{!./src/sql_databases/tutorial001.py!} |
|||
```Python hl_lines="16 17 18 19 20 21 22 23 24" |
|||
{!./src/sql_databases/sql_app/main.py!} |
|||
``` |
|||
|
|||
## Create your **FastAPI** code |
|||
!!! info |
|||
We put the creation of the `SessionLocal()` and handling of the requests in a `try` block. |
|||
|
|||
And then we close it in the `finally` block. |
|||
|
|||
This way we make sure the database session is always closed after the request. Even if there was an exception while processing the request. |
|||
|
|||
Now, finally, here's the standard **FastAPI** code. |
|||
#### About `request.state` |
|||
|
|||
Create your app and path operation function: |
|||
<a href="https://www.starlette.io/requests/#other-state" target="_blank">`request.state` is a property of each Starlette `Request` object</a>, it is there to store arbitrary objects attached to the request itself, like the database session in this case. |
|||
|
|||
```Python hl_lines="59 62 63 64 65" |
|||
{!./src/sql_databases/tutorial001.py!} |
|||
For us in this case, it helps us ensuring a single session/database-connection is used through all the request, and then closed afterwards (in the middleware). |
|||
|
|||
### Create a dependency |
|||
|
|||
To simplify the code, reduce repetition and get better editor support, we will create a dependency that returns this same database session from the request. |
|||
|
|||
And when using the dependency in a path operation function, we declare it with the type `Session` we imported directly from SQLAlchemy. |
|||
|
|||
This will then give us better editor support inside the path operation function, because the editor will know that the `db` parameter is of type `Session`. |
|||
|
|||
```Python hl_lines="28 29" |
|||
{!./src/sql_databases/sql_app/main.py!} |
|||
``` |
|||
|
|||
!!! info "Technical Details" |
|||
The parameter `db` is actually of type `SessionLocal`, but this class (created with `sessionmaker()`) is a "proxy" of a SQLAlchemy `Session`, so, the editor doesn't really know what methods are provided. |
|||
|
|||
But by declaring the type as `Session`, the editor now can know the available methods (`.add()`, `.query()`, `.commit()`, etc) and can provide better support (like completion). The type declaration doesn't affect the actual object. |
|||
|
|||
### Create your **FastAPI** *path operations* |
|||
|
|||
Now, finally, here's the standard **FastAPI** *path operations* code. |
|||
|
|||
```Python hl_lines="32 33 34 35 36 37 40 41 42 43 46 47 48 49 50 51 54 55 56 57 58 61 62 63 64 65" |
|||
{!./src/sql_databases/sql_app/main.py!} |
|||
``` |
|||
|
|||
We are creating the database session before each request, attaching it to the request, and then closing it afterwards. |
|||
@ -226,32 +491,45 @@ Then, in the dependency `get_db()` we are extracting the database session from t |
|||
|
|||
And then we can create the dependency in the path operation function, to get that session directly. |
|||
|
|||
With that, we can just call `get_user` directly from inside of the path operation function and use that session. |
|||
With that, we can just call `crud.get_user` directly from inside of the path operation function and use that session. |
|||
|
|||
Having this 3-step process (middleware, dependency, path operation) in this simple example might seem like an overkill. But imagine if you had 20 or 100 path operations, doing this, you would be reducing a lot of code repetition, and getting better support/checks/completion in all those path operation functions. |
|||
Having this 3-step process (middleware, dependency, path operation) you get better support/checks/completion in all the path operation functions while reducing code repetition. |
|||
|
|||
## Create the path operation function |
|||
!!! tip |
|||
Notice that the values you return are SQLAlchemy models, or lists of SQLAlchemy models. |
|||
|
|||
Here we are using SQLAlchemy code inside of the path operation function, and in turn it will go and communicate with an external database. |
|||
But as all the *path operations* have a `response_model` with Pydantic *models* / schemas using `orm_mode`, the data declared in your Pydantic models will be extracted from them and returned to the client, with all the normal filtering and validation. |
|||
|
|||
!!! tip |
|||
Also notice that there are `response_models` that have standard Python types like `List[schemas.Item]`. |
|||
|
|||
But as the content/parameter of that `List` is a Pydantic *model* with `orm_mode`, the data will be retrieved and returned to the client as normally, without problems. |
|||
|
|||
### About `def` vs `async def` |
|||
|
|||
Here we are using SQLAlchemy code inside of the path operation function, and, in turn, it will go and communicate with an external database. |
|||
|
|||
That could potentially require some "waiting". |
|||
|
|||
But as SQLAlchemy doesn't have compatibility for using `await` directly, as would be with something like: |
|||
|
|||
```Python |
|||
user = await get_user(db_session, user_id=user_id) |
|||
user = await db.query(User).first() |
|||
``` |
|||
|
|||
...and instead we are using: |
|||
|
|||
```Python |
|||
user = get_user(db_session, user_id=user_id) |
|||
user = db.query(User).first() |
|||
``` |
|||
|
|||
Then we should declare the path operation without `async def`, just with a normal `def`: |
|||
Then we should declare the path operation without `async def`, just with a normal `def`, as: |
|||
|
|||
```Python hl_lines="63" |
|||
{!./src/sql_databases/tutorial001.py!} |
|||
```Python hl_lines="2" |
|||
@app.get("/users/{user_id}", response_model=schemas.User) |
|||
def read_user(user_id: int, db: Session = Depends(get_db)): |
|||
db_user = crud.get_user(db, user_id=user_id) |
|||
... |
|||
``` |
|||
|
|||
!!! note "Very Technical Details" |
|||
@ -261,7 +539,45 @@ Then we should declare the path operation without `async def`, just with a norma |
|||
|
|||
Because we are using SQLAlchemy directly and we don't require any kind of plug-in for it to work with **FastAPI**, we could integrate database <abbr title="Automatically updating the database to have any new column we define in our models.">migrations</abbr> with <a href="https://alembic.sqlalchemy.org" target="_blank">Alembic</a> directly. |
|||
|
|||
You would probably want to declare your database and models in a different file or set of files, this would allow Alembic to import it and use it without even needing to have **FastAPI** installed for the migrations. |
|||
And as the code related to SQLAlchemy and the SQLAlchemy models lives in separate independent files, you would even be able to perform the migrations with Alembic without having to install FastAPI, Pydantic, or anything else. |
|||
|
|||
The same way, you would be able to use the same SQLAlchemy models and utilities in other parts of your code that are not related to **FastAPI**. |
|||
|
|||
For example, in a background task worker with <a href="http://www.celeryproject.org/" target="_blank">Celery</a>, <a href="https://python-rq.org/" target="_blank">RQ</a>, or <a href="https://arq-docs.helpmanual.io/" target="_blank">ARQ</a>. |
|||
|
|||
## Review all the files |
|||
|
|||
* `sql_app/__init__.py`: is an empty file. |
|||
|
|||
* `sql_app/database.py`: |
|||
|
|||
```Python hl_lines="" |
|||
{!./src/sql_databases/sql_app/database.py!} |
|||
``` |
|||
|
|||
* `sql_app/models.py`: |
|||
|
|||
```Python hl_lines="" |
|||
{!./src/sql_databases/sql_app/models.py!} |
|||
``` |
|||
|
|||
* `sql_app/schemas.py`: |
|||
|
|||
```Python hl_lines="" |
|||
{!./src/sql_databases/sql_app/schemas.py!} |
|||
``` |
|||
|
|||
* `sql_app/crud.py`: |
|||
|
|||
```Python hl_lines="" |
|||
{!./src/sql_databases/sql_app/crud.py!} |
|||
``` |
|||
|
|||
* `sql_app/main.py`: |
|||
|
|||
```Python hl_lines="" |
|||
{!./src/sql_databases/sql_app/main.py!} |
|||
``` |
|||
|
|||
## Check it |
|||
|
|||
@ -272,12 +588,12 @@ You can copy this code and use it as is. |
|||
In fact, the code shown here is part of the tests. As most of the code in these docs. |
|||
|
|||
|
|||
You can copy it, let's say, to a file `main.py`. |
|||
You can copy it as is. |
|||
|
|||
Then you can run it with Uvicorn: |
|||
|
|||
```bash |
|||
uvicorn main:app --reload |
|||
uvicorn sql_app.main:app --reload |
|||
``` |
|||
|
|||
And then, you can open your browser at <a href="http://127.0.0.1:8000/docs" target="_blank">http://127.0.0.1:8000/docs</a>. |
|||
@ -286,27 +602,6 @@ And you will be able to interact with your **FastAPI** application, reading data |
|||
|
|||
<img src="/img/tutorial/sql-databases/image01.png"> |
|||
|
|||
## Response schema and security |
|||
|
|||
This section has the minimum code to show how it works and how you can integrate SQLAlchemy with FastAPI. |
|||
|
|||
But it is recommended that you also create a response model with Pydantic, as described in the section about <a href="/tutorial/extra-models/" target="_blank">Extra Models</a>. |
|||
|
|||
That way you will document the schema of the responses of your API, and you will be able to limit/filter the returned data. |
|||
|
|||
Limiting the returned data is important for security, as for example, you shouldn't be returning the `hashed_password` to the clients. |
|||
|
|||
That's something that you can improve in this example application, here's the current response data: |
|||
|
|||
```JSON |
|||
{ |
|||
"is_active": true, |
|||
"hashed_password": "notreallyhashed", |
|||
"email": "[email protected]", |
|||
"id": 1 |
|||
} |
|||
``` |
|||
|
|||
## Interact with the database directly |
|||
|
|||
If you want to explore the SQLite database (file) directly, independently of FastAPI, to debug its contents, add tables, columns, records, modify data, etc. you can use <a href="https://sqlitebrowser.org/" target="_blank">DB Browser for SQLite</a>. |
|||
|
|||
@ -1,6 +1,6 @@ |
|||
from starlette.testclient import TestClient |
|||
|
|||
from sql_databases.tutorial001 import app |
|||
from body_nested_models.tutorial009 import app |
|||
|
|||
client = TestClient(app) |
|||
|
|||
@ -8,8 +8,8 @@ openapi_schema = { |
|||
"openapi": "3.0.2", |
|||
"info": {"title": "Fast API", "version": "0.1.0"}, |
|||
"paths": { |
|||
"/users/{user_id}": { |
|||
"get": { |
|||
"/index-weights/": { |
|||
"post": { |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
@ -26,16 +26,20 @@ openapi_schema = { |
|||
}, |
|||
}, |
|||
}, |
|||
"summary": "Read User", |
|||
"operationId": "read_user_users__user_id__get", |
|||
"parameters": [ |
|||
{ |
|||
"required": True, |
|||
"schema": {"title": "User_Id", "type": "integer"}, |
|||
"name": "user_id", |
|||
"in": "path", |
|||
} |
|||
], |
|||
"summary": "Create Index Weights", |
|||
"operationId": "create_index_weights_index-weights__post", |
|||
"requestBody": { |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"title": "Weights", |
|||
"type": "object", |
|||
"additionalProperties": {"type": "number"}, |
|||
} |
|||
} |
|||
}, |
|||
"required": True, |
|||
}, |
|||
} |
|||
} |
|||
}, |
|||
@ -77,12 +81,23 @@ def test_openapi_schema(): |
|||
assert response.json() == openapi_schema |
|||
|
|||
|
|||
def test_first_user(): |
|||
response = client.get("/users/1") |
|||
def test_post_body(): |
|||
data = {"2": 2.2, "3": 3.3} |
|||
response = client.post("/index-weights/", json=data) |
|||
assert response.status_code == 200 |
|||
assert response.json() == data |
|||
|
|||
|
|||
def test_post_invalid_body(): |
|||
data = {"foo": 2.2, "3": 3.3} |
|||
response = client.post("/index-weights/", json=data) |
|||
assert response.status_code == 422 |
|||
assert response.json() == { |
|||
"is_active": True, |
|||
"hashed_password": "notreallyhashed", |
|||
"email": "[email protected]", |
|||
"id": 1, |
|||
"detail": [ |
|||
{ |
|||
"loc": ["body", "weights", "__key__"], |
|||
"msg": "value is not a valid integer", |
|||
"type": "type_error.integer", |
|||
} |
|||
] |
|||
} |
|||
@ -0,0 +1,44 @@ |
|||
from starlette.testclient import TestClient |
|||
|
|||
from extra_models.tutorial005 import app |
|||
|
|||
client = TestClient(app) |
|||
|
|||
openapi_schema = { |
|||
"openapi": "3.0.2", |
|||
"info": {"title": "Fast API", "version": "0.1.0"}, |
|||
"paths": { |
|||
"/keyword-weights/": { |
|||
"get": { |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"title": "Response_Read_Keyword_Weights", |
|||
"type": "object", |
|||
"additionalProperties": {"type": "number"}, |
|||
} |
|||
} |
|||
}, |
|||
} |
|||
}, |
|||
"summary": "Read Keyword Weights", |
|||
"operationId": "read_keyword_weights_keyword-weights__get", |
|||
} |
|||
} |
|||
}, |
|||
} |
|||
|
|||
|
|||
def test_openapi_schema(): |
|||
response = client.get("/openapi.json") |
|||
assert response.status_code == 200 |
|||
assert response.json() == openapi_schema |
|||
|
|||
|
|||
def test_get_items(): |
|||
response = client.get("/keyword-weights/") |
|||
assert response.status_code == 200 |
|||
assert response.json() == {"foo": 2.3, "bar": 3.4} |
|||
@ -0,0 +1,353 @@ |
|||
from starlette.testclient import TestClient |
|||
|
|||
from sql_databases.sql_app.main import app |
|||
|
|||
client = TestClient(app) |
|||
|
|||
openapi_schema = { |
|||
"openapi": "3.0.2", |
|||
"info": {"title": "Fast API", "version": "0.1.0"}, |
|||
"paths": { |
|||
"/users/": { |
|||
"get": { |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"title": "Response_Read_Users", |
|||
"type": "array", |
|||
"items": {"$ref": "#/components/schemas/User"}, |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
"422": { |
|||
"description": "Validation Error", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/HTTPValidationError" |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
}, |
|||
"summary": "Read Users", |
|||
"operationId": "read_users_users__get", |
|||
"parameters": [ |
|||
{ |
|||
"required": False, |
|||
"schema": {"title": "Skip", "type": "integer", "default": 0}, |
|||
"name": "skip", |
|||
"in": "query", |
|||
}, |
|||
{ |
|||
"required": False, |
|||
"schema": {"title": "Limit", "type": "integer", "default": 100}, |
|||
"name": "limit", |
|||
"in": "query", |
|||
}, |
|||
], |
|||
}, |
|||
"post": { |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": {"$ref": "#/components/schemas/User"} |
|||
} |
|||
}, |
|||
}, |
|||
"422": { |
|||
"description": "Validation Error", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/HTTPValidationError" |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
}, |
|||
"summary": "Create User", |
|||
"operationId": "create_user_users__post", |
|||
"requestBody": { |
|||
"content": { |
|||
"application/json": { |
|||
"schema": {"$ref": "#/components/schemas/UserCreate"} |
|||
} |
|||
}, |
|||
"required": True, |
|||
}, |
|||
}, |
|||
}, |
|||
"/users/{user_id}": { |
|||
"get": { |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": {"$ref": "#/components/schemas/User"} |
|||
} |
|||
}, |
|||
}, |
|||
"422": { |
|||
"description": "Validation Error", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/HTTPValidationError" |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
}, |
|||
"summary": "Read User", |
|||
"operationId": "read_user_users__user_id__get", |
|||
"parameters": [ |
|||
{ |
|||
"required": True, |
|||
"schema": {"title": "User_Id", "type": "integer"}, |
|||
"name": "user_id", |
|||
"in": "path", |
|||
} |
|||
], |
|||
} |
|||
}, |
|||
"/users/{user_id}/items/": { |
|||
"post": { |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": {"$ref": "#/components/schemas/Item"} |
|||
} |
|||
}, |
|||
}, |
|||
"422": { |
|||
"description": "Validation Error", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/HTTPValidationError" |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
}, |
|||
"summary": "Create Item For User", |
|||
"operationId": "create_item_for_user_users__user_id__items__post", |
|||
"parameters": [ |
|||
{ |
|||
"required": True, |
|||
"schema": {"title": "User_Id", "type": "integer"}, |
|||
"name": "user_id", |
|||
"in": "path", |
|||
} |
|||
], |
|||
"requestBody": { |
|||
"content": { |
|||
"application/json": { |
|||
"schema": {"$ref": "#/components/schemas/ItemCreate"} |
|||
} |
|||
}, |
|||
"required": True, |
|||
}, |
|||
} |
|||
}, |
|||
"/items/": { |
|||
"get": { |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"title": "Response_Read_Items", |
|||
"type": "array", |
|||
"items": {"$ref": "#/components/schemas/Item"}, |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
"422": { |
|||
"description": "Validation Error", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/HTTPValidationError" |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
}, |
|||
"summary": "Read Items", |
|||
"operationId": "read_items_items__get", |
|||
"parameters": [ |
|||
{ |
|||
"required": False, |
|||
"schema": {"title": "Skip", "type": "integer", "default": 0}, |
|||
"name": "skip", |
|||
"in": "query", |
|||
}, |
|||
{ |
|||
"required": False, |
|||
"schema": {"title": "Limit", "type": "integer", "default": 100}, |
|||
"name": "limit", |
|||
"in": "query", |
|||
}, |
|||
], |
|||
} |
|||
}, |
|||
}, |
|||
"components": { |
|||
"schemas": { |
|||
"ItemCreate": { |
|||
"title": "ItemCreate", |
|||
"required": ["title"], |
|||
"type": "object", |
|||
"properties": { |
|||
"title": {"title": "Title", "type": "string"}, |
|||
"description": {"title": "Description", "type": "string"}, |
|||
}, |
|||
}, |
|||
"Item": { |
|||
"title": "Item", |
|||
"required": ["title", "id", "owner_id"], |
|||
"type": "object", |
|||
"properties": { |
|||
"title": {"title": "Title", "type": "string"}, |
|||
"description": {"title": "Description", "type": "string"}, |
|||
"id": {"title": "Id", "type": "integer"}, |
|||
"owner_id": {"title": "Owner_Id", "type": "integer"}, |
|||
}, |
|||
}, |
|||
"User": { |
|||
"title": "User", |
|||
"required": ["email", "id", "is_active"], |
|||
"type": "object", |
|||
"properties": { |
|||
"email": {"title": "Email", "type": "string"}, |
|||
"id": {"title": "Id", "type": "integer"}, |
|||
"is_active": {"title": "Is_Active", "type": "boolean"}, |
|||
"items": { |
|||
"title": "Items", |
|||
"type": "array", |
|||
"items": {"$ref": "#/components/schemas/Item"}, |
|||
"default": [], |
|||
}, |
|||
}, |
|||
}, |
|||
"UserCreate": { |
|||
"title": "UserCreate", |
|||
"required": ["email", "password"], |
|||
"type": "object", |
|||
"properties": { |
|||
"email": {"title": "Email", "type": "string"}, |
|||
"password": {"title": "Password", "type": "string"}, |
|||
}, |
|||
}, |
|||
"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(): |
|||
response = client.get("/openapi.json") |
|||
assert response.status_code == 200 |
|||
assert response.json() == openapi_schema |
|||
|
|||
|
|||
def test_create_user(): |
|||
test_user = {"email": "[email protected]", "password": "secret"} |
|||
response = client.post("/users/", json=test_user) |
|||
assert response.status_code == 200 |
|||
data = response.json() |
|||
assert test_user["email"] == data["email"] |
|||
assert "id" in data |
|||
response = client.post("/users/", json=test_user) |
|||
assert response.status_code == 400 |
|||
|
|||
|
|||
def test_get_user(): |
|||
response = client.get("/users/1") |
|||
assert response.status_code == 200 |
|||
data = response.json() |
|||
assert "email" in data |
|||
assert "id" in data |
|||
|
|||
|
|||
def test_inexistent_user(): |
|||
response = client.get("/users/999") |
|||
assert response.status_code == 404 |
|||
|
|||
|
|||
def test_get_users(): |
|||
response = client.get("/users/") |
|||
assert response.status_code == 200 |
|||
data = response.json() |
|||
assert "email" in data[0] |
|||
assert "id" in data[0] |
|||
|
|||
|
|||
def test_create_item(): |
|||
item = {"title": "Foo", "description": "Something that fights"} |
|||
response = client.post("/users/1/items/", json=item) |
|||
assert response.status_code == 200 |
|||
item_data = response.json() |
|||
assert item["title"] == item_data["title"] |
|||
assert item["description"] == item_data["description"] |
|||
assert "id" in item_data |
|||
assert "owner_id" in item_data |
|||
response = client.get("/users/1") |
|||
assert response.status_code == 200 |
|||
user_data = response.json() |
|||
item_to_check = [it for it in user_data["items"] if it["id"] == item_data["id"]][0] |
|||
assert item_to_check["title"] == item["title"] |
|||
assert item_to_check["description"] == item["description"] |
|||
response = client.get("/users/1") |
|||
assert response.status_code == 200 |
|||
user_data = response.json() |
|||
item_to_check = [it for it in user_data["items"] if it["id"] == item_data["id"]][0] |
|||
assert item_to_check["title"] == item["title"] |
|||
assert item_to_check["description"] == item["description"] |
|||
|
|||
|
|||
def test_read_items(): |
|||
response = client.get("/items/") |
|||
assert response.status_code == 200 |
|||
data = response.json() |
|||
assert data |
|||
first_item = data[0] |
|||
assert "title" in first_item |
|||
assert "description" in first_item |
|||
Loading…
Reference in new issue