sql_app/main.py이전에 만든 다른 모든 부분을 통합하고 사용하겠습니다.
데이터베이스 테이블 생성

from typing import List

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

- 일반적으로 알렘빅을 사용하여 데이터베이스를 초기화 및 생성한다.
- 또한 migrations(main job)에 사용 가능하다
- 마이그레이션은 SQLAlchemy모델의 구조를 변경할 때마다 필요한 일련의 단계로 새 속성을 추가하는 등 데이터베이스에서 이러한 변경 사항을 복제하고 새 열이나 새 테이블 등을 추가한다.

> dependency(종속성) 생성
- 파일 `SessionLocal`에서 만든 클래스 `sql_app/database.py`를 사용하여  dependency을 만든다.
- 요청마다 독립적인 데이터베이스 세션/연결이 있어야 하며 모든 요청을 통해 동일한 세션을 사용한 다음 요청이 완료된 후 종료한다.
- 그 후 다음 요청을 위한 새 세션이 생성된다.
- 이를 위해  dependency `yield`에 대한 섹션에서 새  dependency을 생성한다.
- dependency은 새로운 SQLAlchemy `SessionLocal`을 생성하고 요청이 완료되면 종료한다.

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

SessionLocal()요청 생성 및 처리를 try블록 에 넣습니다 . 그런 다음 finally블록에서 닫습니다. 이렇게 하면 요청 후에 데이터베이스 세션이 항상 닫힙니다. 요청을 처리하는 동안 예외가 발생한 경우에도 마찬가지입니다. 그러나 종료 코드(다음)에서 다른 예외를 발생시킬 수 없습니다 yield. 및 종속성 에서 자세히 보기yield``HTTPException

마지막으로 표준 FastAPI 경로적업코드가 있다.

• dependency에서 각 요청 전에 데이터베이스 세션을 yield에 만든 후 닫는다.
• 경로 작업 함수에 필요한 dependency를 생성하여 해당 섹션 직접 가져올 수 있다.
• 이를 통해 경로 연산 함수 crud.get_user 내부에서 직접 호출하고 해당 세션을 사용할 수 있다.