[FastAPI/Python] FastAPI와 데이터 베이스

2022. 8. 8. 22:05· 개발/오늘 배운 지식
목차
  1. Relational vs NoSQL 데이터베이스
  2. 관계형 DB
  3. NoSQL DB
  4. 어떤 것을 선택할까?
  5. 관계형 DB를 사용하는 경우
  6. 테이블 스키마 생성하기
  7. 데이터베이스 연결하기
  8. Insert 쿼리 만들기
  9. Select 쿼리
  10. Update, Delete 쿼리 만들기
  11. Relationship 추가하기
  12. Alembic을 활용해서 DB 마이그레이션 시스템 세팅하기
반응형

Chapter 6: 데이터 베이스와 비동기 ORMs

“Building Data Science Applications with FastAPI”라는 책을 읽고 정리한 내용입니다.

REST API를 만드는 주된 이유: 데이터를 읽고 쓰기 위함.

하지만, 아직 데이터베이스에 대해서는 책에서 다루지 않았음.

  • ORM: Object-Relational Mapping

Relational vs NoSQL 데이터베이스

  • 데이터베이스는 체계적으로 데이터를 저장하고, 데이터의 무결성을 유지하며, query language를 통해 필요할 때 데이터를 찾을 수 있도록 해줘야 한다.
  • Relational Database(관계형 데이터베이스): SQL query language
  • NoSQL: Relational DB의 반대 케이스
  • 필요에 따라 적절한 기술 선택하는 것이 필요하다.

관계형 DB

  • SQL 쿼리 언어와 함께 사용
  • relational model을 구현한다
    • entity or object of application is stored in tables
    • 모든 것이 table 형태로 저장
    • 각 table의 column은 entity의 attribute를 나타냄
    • table의 row는 하나의 entity를 뜻한다.
    • 각 table들은 서로 relationship을 가진다.
      • 이 관계를 만드는 주된 동기는 avoid duplication
  • 보통, 관계형 DB의 각 row는 primary key라는 identifier를 가진다.
    • unique in the table → uniquely identify this row
    • 다른 table이 이 row를 찾는 데 사용한다면 foreign key가 되는 것임
      • 즉, 다른 table의 primary key를 현재 table에 저장하고 있는 것.
  • Schema가 너무 복잡해지면, foreign key로 찾아서 join 하는 것이 오히려 더 비효율 적일 수 있다.

NoSQL DB

  • Not relational DB ⇒ fallback to NoSQL
    • Key-Value stores: Redis
    • Graph DB: Neo4j
    • Document-oriented DB: MongoDB
  • 현재 이 책에서 NoSQL을 지칭할 때는 위의 세 가지 중 마지막인 Document-oriented DB를 지칭함.
  • 문서 지향 DB는 하나의 문서 내에 모든 정보를 저장한다는 컨셉이다
    • 따라서, join query를 더 적게 수행하고, join이 어려움.
  • documents들은 collections에 저장된다.
    • 각 문서는 서로 다른 attribute를 가지고 있을 수 있다(일관성이 보장되지 않음)
  • 예시: 게시물에 달리는 댓글을 저장할 때, 댓글 컬렉션을 따로 구별하지 않고, 게시물 document 내에 comments list를 저장하는 것.
    • 거대한 스케일, 덜 구조적인 데이터를 저장할 때, 적합하다.
      • Social networks

어떤 것을 선택할까?

  • 관계형 DB
    • 구조적으로 데이터 저장 가능
    • 데이터가 일관성을 유지할 수 있음
    • 하지만, 스키마를 정의할 때, 세심하게 진행해야 함.
    • schema 수정이 힘들다
  • Document-oriented DB
    • 스키마를 정의할 필요가 없다.
    • 아직 프로젝트가 성숙하지 않았을 때, 유연한 변경이 가능
    • 데이터 일관성 유지가 힘들다.
  • 작거나 중간 정도의 앱에서는 두 가지 모두 좋은 성능을 보여준다.

관계형 DB를 사용하는 경우

  • SQLAlchemy 사용
  • Encode를 활용해서 비동기 통신 제어
    • 여기서는 sqlite를 사용
    • production level에서는 mysql, postgre-sql 등을 사용 가능하다

먼저, 라이브러리를 가상 환경에 설치하자.

pip install databases[sqlite]

테이블 스키마 생성하기

metadata = sqlalchemy.MetaData()

posts = sqlalchemy.Table(

    "posts",

    metadata,

    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True, autoincrement=True),

    sqlalchemy.Column("publication_date", sqlalchemy.DateTime(), nullable=False),

    sqlalchemy.Column("title", sqlalchemy.String(length=255), nullable=False),

    sqlalchemy.Column("content", sqlalchemy.Text(), nullable=False),

)

먼저 메타데이터 객체를 생성한다. DB 자체의 정보를 담고 있다. 전체 프로젝트에서 딱 한번 생성 후, 같은 객체에 접근하게 된다.

sqlalchemy의 Table 클래스를 이용하여 Table을 생성하고, Column 클래스를 이용하여 entity의 attribute를 생성할 수 있다.

데이터베이스 연결하기

DATABASE_URL = "sqlite:///chapter6_sqlalchemy.db"

database = Database(DATABASE_URL)

sqlalchemy_engine = sqlalchemy.create_engine(DATABASE_URL)

여기서 DATABASE_URL은 postgresql+pg8000://dbuser:kx%25jj5%2Fg**@pghost10**/appdb 처럼 데이터베이스 엔진, auth 정보와 호스트 이름 등이 적혀있다. sqlite는 db file의 path만 간단히 입력해두면 된다.

def get_database() -> Database:

    return database

위의 코드는 DB 인스턴스를 받아오는 Dependency를 구현한 것이다. 이렇게 구현하는 것의 장점은 Unit Test가 쉬워지기 때문이다.

@app.on_event("startup")
async def startup():

    await database.connect()

    metadata.create_all(sqlalchemy_engine)

@app.on_event("shutdown")
async def shutdown():

    await database.disconnect()

그리고 FastAPI는 startup과 shutdown 이벤트를 위한 데코레이터를 제공한다.

Insert 쿼리 만들기

@app.post("/posts", response_model=PostDB, status_code=status.HTTP_201_CREATED)
async def create_post(

    post: PostCreate, database: Database = Depends(get_database)

) -> PostDB:

    insert_query = posts.insert().values(post.dict())

    post_id = await database.execute(insert_query)

    post_db = await get_post_or_404(post_id, database)

    return post_db
  • SQLAlchemy에서 제공하는 메서드로 쿼리를 만들면, DB를 다른 엔진으로 바꿔도 코드 변경 없이 사용할 수 있다
  • 그리고 쿼리가 posts라는 Table instance를 활용하였기 때문에 SQLAlchemy가 바로 어떤 테이블에 쿼리를 만들었는지 알 수 있다
  • post.dict()를 통해 편리하게 값들을 설정할 수 있다. 이렇게 하기 위해서는 Pydantic Model과 Table schema가 통일성 있게 유지되어야 한다.
  • await 구문을 통해서 비동기적으로 실행된다.
  • 마지막으로 get_post_or_404로 디비에 올바르게 저장되었는지 확인한다.

Select 쿼리

가장 기본이 되는 select 쿼리문을 만들어보자

하나의 객체만 가져오거나, 리스트 객체를 받아온다.

@app.get("/posts")
async def list_posts(

    pagination: Tuple[int, int] = Depends(pagination),

    database: Database = Depends(get_database),

) -> List[PostDB]:

    skip, limit = pagination

    select_query = posts.select().offset(skip).limit(limit)

    rows = await database.fetch_all(select_query)

    results = [PostDB(**row) for row in rows]

    return results

@app.get("/posts/{id}", response_model=PostDB)

async def get_post(post: PostDB = Depends(get_post_or_404)) -> PostDB:

    return post

get_post_or_404처럼 하나의 객체를 찾는 경우가 잦기 때문에 의존성을 구현해서 재사용성을 높이자.

async def get_post_or_404(

    id: int, database: Database = Depends(get_database)

) -> PostDB:

    select_query = posts.select().where(posts.c.id == id)

    raw_post = await database.fetch_one(select_query)

    if raw_post is None:

        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)

    return PostDB(**raw_post)

WHERE절을 이용한다면 우리가 원하는 조건의 데이터를 필터링할 수 있다. column에는 posts.c.id 로 접근할 수 있다.

💡 *DEPENDENCIES ARE LIKE FUNCTIONS*

Update, Delete 쿼리 만들기

@app.patch("/posts/{id}", response_model=PostDB)
async def update_post(

    post_update: PostPartialUpdate,

    post: PostDB = Depends(get_post_or_404),

    database: Database = Depends(get_database),

) -> PostDB:

    update_query = (

        posts.update()

        .where(posts.c.id == post.id)

        .values(post_update.dict(exclude_unset=True))

    )

    await database.execute(update_query)

    post_db = await get_post_or_404(post.id, database)

    return post_db

@app.delete("/posts/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_post(

    post: PostDB = Depends(get_post_or_404), database: Database = Depends(get_database)

):

    delete_query = posts.delete().where(posts.c.id == post.id)

    await database.execute(delete_query)

두 쿼리문 모두 특정 id를 가지는 객체에만 접근해야 하므로 WHERE 절을 활용한다. 그 외에는 거의 비슷하다.

Relationship 추가하기

comments = sqlalchemy.Table(

    "comments",

    metadata,

    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True, autoincrement=True),

    sqlalchemy.Column(

        "post_id", sqlalchemy.ForeignKey("posts.id", ondelete="CASCADE"), nullable=False

    ),

    sqlalchemy.Column("publication_date", sqlalchemy.DateTime(), nullable=False),

    sqlalchemy.Column("content", sqlalchemy.Text(), nullable=False),

)
@app.post("/comments", response_model=CommentDB, status_code=status.HTTP_201_CREATED)

async def create_comment(

    comment: CommentCreate, database: Database = Depends(get_database)

) -> CommentDB:

    select_post_query = posts.select().where(posts.c.id == comment.post_id)

    post = await database.fetch_one(select_post_query)

    if post is None:

        raise HTTPException(

            status_code=status.HTTP_400_BAD_REQUEST, detail=f"Post {id} does not exist"

        )

    insert_query = comments.insert().values(comment.dict())

    comment_id = await database.execute(insert_query)

    select_query = comments.select().where(comments.c.id == comment_id)

    raw_comment = cast(Mapping, await database.fetch_one(select_query))

    return CommentDB(**raw_comment)

하나의 쿼리에서 연관된 다른 테이블의 데이터를 가져오는 코드

async def get_post_or_404(

    id: int, database: Database = Depends(get_database)

) -> PostPublic:

    select_post_query = posts.select().where(posts.c.id == id)

    raw_post = await database.fetch_one(select_post_query)

    if raw_post is None:

        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)

    select_post_comments_query = comments.select().where(comments.c.post_id == id)

    raw_comments = await database.fetch_all(select_post_comments_query)

    comments_list = [CommentDB(**comment) for comment in raw_comments]

    return PostPublic(**raw_post, comments=comments_list)

Alembic을 활용해서 DB 마이그레이션 시스템 세팅하기

pip install alembic로 설치 후 프로젝트 루트 경로에서 alembic init alembic 명령어를 실행하자. 깃헙에 올릴 때, 함께 업로드가 되어야 한다. Alembic에 DB의 메타데이터를 제공해주면 마이그레이션 스크립트를 자동 생성해준다. 설정 변경이 필수적임.

  • DB 스키마를 업데이트하는 마이그레이션은 극히 신중하게 수행되어야 한다.
반응형
저작자표시 비영리 동일조건 (새창열림)

'개발 > 오늘 배운 지식' 카테고리의 다른 글

[FastAPI/Python] FastAPI에서의 인증과 보안  (0) 2022.08.11
[FastAPI/Python] FastAPI를 Tortoise ORM과 MongoDB와 연동하기  (2) 2022.08.09
[FastAPI, Python] FastAPI의 의존성 구현과 주입 방법  (0) 2022.08.06
[react/jest] jest에서 konva를 인식하지 못하는 경우 해결 방법.  (0) 2021.11.27
[django/python] 장고와 pylint 충돌  (0) 2021.11.02
  1. Relational vs NoSQL 데이터베이스
  2. 관계형 DB
  3. NoSQL DB
  4. 어떤 것을 선택할까?
  5. 관계형 DB를 사용하는 경우
  6. 테이블 스키마 생성하기
  7. 데이터베이스 연결하기
  8. Insert 쿼리 만들기
  9. Select 쿼리
  10. Update, Delete 쿼리 만들기
  11. Relationship 추가하기
  12. Alembic을 활용해서 DB 마이그레이션 시스템 세팅하기
'개발/오늘 배운 지식' 카테고리의 다른 글
  • [FastAPI/Python] FastAPI에서의 인증과 보안
  • [FastAPI/Python] FastAPI를 Tortoise ORM과 MongoDB와 연동하기
  • [FastAPI, Python] FastAPI의 의존성 구현과 주입 방법
  • [react/jest] jest에서 konva를 인식하지 못하는 경우 해결 방법.
Woogie2
Woogie2
창업, 개발, AI
반응형
Woogie2
Dev In Seoul
Woogie2
전체
오늘
어제
  • Dev in Seoul (47)
    • 개발 (39)
      • 알고리즘 (5)
      • 오늘 배운 지식 (24)
      • 학교 수업 (9)
    • 경험, 생각, 일상 (1)
      • 스타트업 (0)
    • 학교 (2)
    • Machine Learning (5)
      • AI (4)
      • 논문 리뷰 (1)

인기 글

최근 글

최근 댓글

hELLO · Designed By 정상우.v4.2.2
Woogie2
[FastAPI/Python] FastAPI와 데이터 베이스
상단으로

티스토리툴바

개인정보

  • 티스토리 홈
  • 포럼
  • 로그인

단축키

내 블로그

내 블로그 - 관리자 홈 전환
Q
Q
새 글 쓰기
W
W

블로그 게시글

글 수정 (권한 있는 경우)
E
E
댓글 영역으로 이동
C
C

모든 영역

이 페이지의 URL 복사
S
S
맨 위로 이동
T
T
티스토리 홈 이동
H
H
단축키 안내
Shift + /
⇧ + /

* 단축키는 한글/영문 대소문자로 이용 가능하며, 티스토리 기본 도메인에서만 동작합니다.