Metadata

Code

.
├── api
│   ├── __init__.py
│   ├── items.py
│   └── users.py
├── config.py
├── db.py
├── dependencies.py
├── __init__.py
├── main.py
└── models.py

app/__init__.py

app/models.py

from typing import List, Optional

from sqlmodel import Field, Relationship, SQLModel


class ItemBase(SQLModel):
    title: str
    description: Optional[str] = None


class Item(ItemBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    user_id: int = Field(foreign_key="user.id")

    user: Optional["User"] = Relationship(back_populates="items")


class UserBase(SQLModel):
    username: str


class User(UserBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    is_active: bool = True
    items: List[Item] = Relationship(back_populates="user")


class UserWithItems(UserBase):
    id: int
    is_active: bool
    items: List[Item]

app/db.py

from sqlmodel import create_engine

sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"

connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, echo=True, connect_args=connect_args)

app/dependencies.py

import secrets

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlmodel import Session

from .config import secret_token
from .db import engine

bearer_scheme = HTTPBearer()


def authenticate_token(
    credentials: HTTPAuthorizationCredentials = Depends(bearer_scheme),
) -> str:
    if not secrets.compare_digest(credentials.credentials, secret_token):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials"
        )
    return credentials.credentials


def get_session():
    with Session(engine) as session:
        yield session

app/main.py

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from sqlmodel import SQLModel

from . import db, models
from .api import items, users


def create_db_and_tables():
    assert models, "Models should be imported so SQLModel has them registered"
    SQLModel.metadata.create_all(db.engine)


app = FastAPI(
    title="My Super App",
    version="0.3.0",
)

origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",
    "https://fastapiworkshop.yaquelinehoyos.com",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


app.include_router(users.router, tags=["users"])
app.include_router(items.router, tags=["items"])


@app.on_event("startup")
def startup_event():
    create_db_and_tables()

app/config.py

# Don't do this in production, read this token from an environment variable
secret_token = "supersecret"

app/api/__init__.py

app/api/users.py

from typing import List

from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import Session, select

from ..dependencies import authenticate_token, get_session
from ..models import User, UserBase, UserWithItems

router = APIRouter()


@router.post(
    "/users/",
    response_model=User,
    dependencies=[Depends(authenticate_token)],
)
def create_user(user: UserBase, session: Session = Depends(get_session)):
    """
    Create a new user.
    """
    db_user = session.exec(select(User).where(User.username == user.username)).first()
    if db_user:
        raise HTTPException(status_code=400, detail="Username already registered")
    db_user = User.from_orm(user)
    session.add(db_user)
    session.commit()
    session.refresh(db_user)
    return db_user


@router.get("/users/", response_model=List[User])
def read_users(
    offset: int = 0, limit: int = 100, session: Session = Depends(get_session)
):
    """
    Read users.

    Doesn't require authentication.
    """
    users = session.exec(select(User).offset(offset).limit(limit)).all()
    return users


@router.get("/users/{user_id}", response_model=UserWithItems)
def read_user(user_id: int, session: Session = Depends(get_session)):
    """
    Read the data for a specific user.

    Doesn't require authentication.
    """
    db_user = session.exec(select(User).where(User.id == user_id)).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

app/api/items.py

from typing import List

from fastapi import APIRouter, Depends
from sqlmodel import Session, select

from ..dependencies import authenticate_token, get_session
from ..models import Item, ItemBase

router = APIRouter()


@router.post(
    "/users/{user_id}/items/",
    response_model=Item,
    dependencies=[Depends(authenticate_token)],
)
def create_item_for_user(
    user_id: int, item: ItemBase, session: Session = Depends(get_session)
):
    """
    Create an item for a specific user.
    """
    db_item = Item.from_orm(item, update={"user_id": user_id})
    session.add(db_item)
    session.commit()
    session.refresh(db_item)
    return db_item


@router.get("/items/", response_model=List[Item])
def read_items(
    offset: int = 0, limit: int = 100, session: Session = Depends(get_session)
):
    """
    Read all the items. Doesn't need authentication.
    """
    items = session.exec(select(Item).offset(offset).limit(limit)).all()
    return items

Server

Run the live server:

$ uvicorn app.main:app --reload

<span style="color: green;">INFO</span>:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
<span style="color: green;">INFO</span>:     Started reloader process [28720]
<span style="color: green;">INFO</span>:     Started server process [28722]
<span style="color: green;">INFO</span>:     Waiting for application startup.
<span style="color: green;">INFO</span>:     Application startup complete.

API docs

Now you can open the API docs UI at: http://127.0.0.1:8000/docs.

Admin Dashboard

You can go to the admin dashboard (created for this workshop) to interact with your API: https://fastapiworkshop.yaquelinehoyos.com.