[FastAPI] SQL (Reational) Databases

FastAPI doesn't require you to use a SQL (relational) database.
But you can use any relational database that you want.
Here we'll see an example using SQLAlchemy.

File structure

.
└── sql_app
├── init.py
├── crud.py
├── database.py
├── main.py
├── models.py
└── schemas.py

init.py는 빈 file이다, 그러나 이것은 모든 module들이 있는 sql_app이 패키지임을 Python에게 말해준다.

Install SQLAlchemy

pip install sqlalchemy

Create the SQLAlchemy parts

Let's refer to the file sql_app/database.py

Import the SQLAlchemy parts

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

// Create a database URL for SQLAlchemy
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"


engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

이 예시에서, SQLite database랑 연결한다.
(opening a file with the SQLite database).
이 파일은 sql_app.db 파일의 같은 directory 안에 위치하게 될 것이다.
이것이 마지막 부분이 ./sql_app.db인 이유이다.

만약 PostgreSQL database를 사용한다면, 아래와 같게 된다:

SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

Create the SQLAchemy engine

The first step is to create a SQLAlchemy "engine".

We will later use this engine in other places.

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

The argument:

connect_args={"check_same_thread": False}

...is needed only for SQLite. It's not needed for other databases.

Create a SessionLocal class

SessionLocal callse의 각 instance는 database session이 된다. class 자체로는 database session이 아니다.

하지만 SessionLocal class의 instance를 만들고 나면, 이 instance는 실제 database session이 된다.

SQLAlchemy로부터 importing 하는 Session으로부터 구별하기 위해 SessionLocal이라고 이름을 지었다.
나중에 Session을 사용하게 될 것이다.

SessionLocal class를 만들기 위해, sessionmaker 함수를 사용한다:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)

// 이 부분
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

Create a Base class

class를 return 하는 declare_base() function을 사용할 것이다.
후에 database models or classes의 각각을 만들기 위해 이 class로부터 상속받을 것이다:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

// 이 부분
Base = declarative_base()

Create the database models

이제 sql_app/models.py을 보자

Create SQLAchemy models from the Base class

SQLAlchemy mdoel을 만들기 전에 만들어둔 Base class를 사용할 것이다.

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).

Create classes that inherit from it.

These classes are the SQLAlchemy models.

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)
    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)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

__tablename__ attribute SQLAlchemy에게 각 모델에 대한 database에서 사용할 table 이름에 대해 알려준다.

Create model attributes/columns

Now create all the model (class) attributes.

이러한 속성의 각각은 연관된 database table의 column을 나타낸다.
기본값으로 SQLAlchemy로부터 column을 사용한다.
그리고 argument로 database에 type을 정의해 Integer, String, and Boolean으로서 SQLAlchemy에 class type을 전달한다.

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)
    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)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

Create the relationships

Now create the relationships.

For this, we use relationship provided by SQLAlchemy ORM.

This will become, more or less, a "magic" attribute that will contain the values from other tables related to this one.

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)
    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)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))
	// 이 부분
    owner = relationship("User", back_populates="items")

my_user.item로 User의 items 속성에 접근할 때, 이것은 users table의 이 record로 pointing 하는 foreign key를 가진 Item SQLAlchemy models의 list를 갖게 된다.

my_user.items에 접근할 때, SQLAlchemy는 실제로 가고, items table의 database로부터 items를 fetch하고, 여기에 채운다.

그리고 Item의 owner 속성에 접근할 때, 이것은 usesr table로부터 User SQLAlchemy model을 포함할 것이다. 이것은 users table로부터 record를 가져오기 위해 이것의 foriegn key를 알려주는 attribute/column인 owner_id를 사용할 것이다.

Create the Pydantic models

Now let's check the file sql_app/schemas.py.

Tip

• SQLAlchemy models과 Pydantic models 사이에서 혼동되는 것을 피하기 위해, 우리는 SQLAlchemy models가 들어간 models.py file을 가지게 될 것이고, Pydantic models가 들어간 schemas.py file을 갖게 될 것이다.
• 이러한 Pydantic models는 "schema"(a valid data shape)를 정의한다

Create initial Pydantic models / schemas

data를 만들고 읽는 동안 common attributes(or let's say "schemas")를 가지기 위해 ItemBase and UserBase Pydantic models을 만든다.

그리고 이것들로부터 상속하는 ItemCreate와 UserCreate를 만들고, 생성에 필요한 추가적인 data(속성)를 더한다.

그래서, user는 만들 때 password를 가지고 있다.

하지만 보안을 위해, password는 다른 Pydantic models에 들어가지 않는다. 예를 들어, 이것은 API로부터 user를 읽어올 때는 보내지지 않는다.

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: str | None = 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

SQLAlchemy style and Pydantic style

SQLAlchemy models는 =를 사용하여 attributes를 정의한다, 그리고 Column에 parameter로 type을 전달한다:

name = Column(String)

하지만 Pydantic models는 :을 사용하여 type을 선언한다:

name: str

Create Pydantic models / schemas for reading / returning

data reading, API로부터 returning할 때 사용될 Pydantic moedels를 만들어 보자.
예를 들어, item을 만들기 전에, 우리는 그것에 할당된 ID가 무엇인지 모른다, 하지만 읽을 때 (API로부터 returning 했을 때) 이미 ID를 알게 된다.

같은 방식으로, user를 읽을 때, 이 user에 속하는 item을 포함하는 items를 선언할 수 있다.

이러한 items의 ID들 뿐만 아니라, items를 읽기 위한 Pydantic models 안에서 정의된 모든 data이다.

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: str | None = 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

Tip:

• User를 인식하여, user를 읽어올 때 사용되는 Pydantic model은 password를 포함하지 않는다.

Use Pydantic's orm_mode

Now, Pydantic models에서 Item and User을 읽기 위해, 내부 Config class를 추가한다.
이 Config class는 Pydantic에 configuration을 제공하기 위해 사용된다.

Config class에서, attribute orm_mode True로 set 한다.

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: str | None = 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

Tip:

• =를 사용한다, orm_mode = True
• :를 사용하지 않는다.
• config value이기 때문이다.

이 방식으로, dict로부터 id value 만을 가져오기 위해 쓰인다:

id = data["id]

이것은 또한 속성으로부터 가져올 때 사용된다:

id = data.id

그리고 이것과 함께, Pydantic model은 ORMs와 호환되며, 그리고 내 Path operations에서 response_model argument에서 사용될 수 있다.

database model을 return 할 수 있고 이것으로부터 data를 읽을 수 있을 것이다.

Technical Details about ORM mode

SQLAlchemy and many others are by default "lazy loading"

이것은 예를 들어, data를 포함한 속성에 접근하려고 시도하지 않는 한 database로부터 relationships에 대한 data를 fetch 해오지 않는다는 것이다.

예를 들어, items 속성에 접근할 때:

current_user.items

would make SQLAlchemy go to the items table and get the items for this user, but not before.

orm_mode 없이, path operation으로부터 만약 SQLAlchemy model을 return한다면, relationship data를 포함하고 있지 않을 것이다.

Pydantic models 안에 그것의 relationship들을 선언해놓았더라도 말이다.

하지만 ORM mode로는, Pydantic 그 자체로 속성으로부터 필요한 data에 접근할 수 있다(dict로 가정하는 대신), return 하고 싶은 특정 data를 선언할 수 있고 이것은 갈 수 있고, ORM으로부터 얻을 수 있다.

CRUD utils

file sql_app/crud.py를 보자
이 파일에서는 database 안에서 data와 상호작용하기 위해 재사용할 수 있는 함수를 가질 수 있다.
CRUD: Create, Read, Update, and Delete

Read data

Import Session from sqlalchemy.orm, 이것은 db parameter의 type을 선언하는 것을 허용하게 하고 더 나은 type check와 함수의 completion을 제공한다.

Import medels (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 multiple items.

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

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 unit tests 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).

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

Info:

• Pydantic v1에서는 method는 .dict()로 호출되었다, 이것은 Pydantic v2에서 충돌되어서 (여전히 지원되지만), .model_dump()로 rename 되었다.

Tip:

• User에 대한 SQLAlchemy model은 password의 안전한 hash된 버전의 hashed_password를 포함한다.
• 하지만 API client가 제공하는 것은 original password로, 이것을 발췌하고 applicatiion에서 hashed password로 재생성할 필요가 있다.
• 그리고 hashed_password argument value를 저장하기 위해 전달한다.

Warning:

• 이 예시에서는 보안 X, password is not hashed.
• 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:

• Item에 keyword argument의 각각을 전달하고 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

이제 file sql_app/main.py
전에 만든 다른 부분을 포함하고 사용해보자

Create the database tables

매우 간단한 방식:

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

// 이 부분
models.Base.metadata.create_all(bind=engine)

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@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

Alembic Note

Normally you would probably initailize your database (create tables, etc) with Alembic.
And you would also use Alembic for "migrations" (that's its main job).
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.

You can find an example of Alembic in a
FastAPI project in the templates from Project Generation - Template. Specifically in the alembic directory in the source code.

Create a dependency

이제 dependency를 만들기 위해 sql_app/database.py 파일에서 만든 class인 SessionLocal class를 사용한다.

요청마다 독립적인 database session/connection (SessionLocal)을 가져야만 하고, 모든 요청과 요청을 통해 같은 session을 사용하고 요청들이 끝난 후에 close 해야만 한다.

이러기 위해, 우리는 이전에 Dependencies with yield에 대한 section에서 설명된 yield라는 new dependency를 만들어야 할 것이다.

우리의 dependency는 single request에서 사용될 new SQLAlchemy SessionLocal을 만들 것이다, 그리고 request가 끝나고 나면 close한다.

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

// 이 부분
# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@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

Info:

• SessionLocal()의 생성과 requests의 조작을 try block에 넣었다.
• 그리고 finally block에서 close 한다.
• 이 방식은 detabase session이 항상 request 후에 close 된다는 것을 확실하게 한다.
• 심지어 request 처리하는 동안 예외가 발생하더라도
• 하지만 exit code 이후에 다른 exception을 발생시킬 수는 없습니다.

그리고, path operation function의 dependency를 사용할 때, 우리는 SQLAlchemy로부터 직접 import한 Session type을 정의한다.

이것은 우리에게 path operation function 내부의 더 나은 editor를 지원한다, 왜냐하면 editor는 dp parameter가 type Session인 것을 알게 하기 때문이다:

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@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

Create your FastAPI path operations

이제, 최종적으로, 기준 FastAPI path operations code가 있다.

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@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

우리는 yield dependency 안에서 각각의 request 전에 database session을 만든다, 그리고 후에 이것을 닫는다.
그리고 우리는 session을 directly하게 얻어오기 위해 paht operaton function 안에서 요청된 dependency를 만들 수 있다.
이것과 함꼐, 우리는 path operation function의 내부로부터 직접적으로 crud.get_user를 호출할 수 있고 session을 사용할 수 있다.

Tip:

• Notice that return하는 값은 SQLAlchemy modelㄷ 또는 그 list이다.
• 하지만 모든 path operations는 orm_mode를 사용하여 Pydantic models / schemas를 가진 response_model을 가지며, Pydantic models 안에서 선언된 data는 모든 normal filtering과 validation과 함께 그것들로부터 추출되고 client에게 return 될 것이다.

Tip:

• 또한 여기엔 List[schemas.Item] 같은 기준 Python type을 가진 response_models가 있다.
• 하지만 그 List의 content/parameter는 orm_mode인 Pydantic model이며, data는 문제 없이 정상적으로 검색되어 클라이언트에 반환될 것이다.

About def vs async def

여기엔 path operation function의 내부와 dependency의 안에서 SQLAlchemy를 사용하고 결국엔 이것을 보내고 외부의 database와 communicate 할 것이다.
이것은 잠재적으로 "waiting"을 요구할 수 있다.
그러나 SQLAlchemy는 직접적으로 waiting을 사용할 수 있는 호환성이 없다:

user = await db.query(User).first()

대신 이렇게 사용한다:

user = db.query(User).first()

그리고 우리는 async def 없이 path operation function과 dependency를 선언해야만 한다, 그저 normal def만 사용해서:

@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)
    ...

Info:

• 만약 비동기적으로 연관된 database에 연결해야 한다면, see Async SQL (Relational) Databases

Very Technical Details:

• If you are curious and have a deep technical knowledge, you can check the very technical details of how this async def vs def is handled in the Async docs.

Migrations

SQLAlchemy를 직접적으로 사용하고 FastAPI와 함께 작동하기 위해 어떠한 plug-in도 필요하지 않기 때문에, we could integrate database migrations with Alembic directly.

And as the code related to SQLAlchemy and the SQLAlchemy models lives in separates independent files, you would even be able to perform the migrations with Alembiv 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 Celery, RQ, or ARQ

Review all the files

Remember you should have a directory named my_super_project that contains a sub-directory called sql_app.

sql_app should have the following files:

• sql_app/__init__.py: is an empty file.
• sql_app/database.py:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

• sql_app/models.py

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)
    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)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

• sql_app/schemas.py:

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: str | None = 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

• sql_app/crud.py:

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

• sql_app/main.py:

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@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

Check it

You can copy this code and use it as is.
Then you can run it with Uvicorn:

uvicorn sql_app.main:app --reload

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

And then, you can open your browser at http://127.0.0.1:8000/docs.

And you will be able to interact with your FastAPI application, reading data from a real database:
...