[Alembic] - Tutorial

* 목차

 - Intro

 - 환경 구성

 - 예제 코드 작성

   1. local_main.py

   2. server_main.py

 - Init Alembic

   1. First Migration

   2. Table 추가, 변경 후 migration

   3. Migration 이전 version으로 rollback

 - Outro

 

---

Intro

Alembic은 Python용 SQLAlchemy 데이터베이스 툴킷과 함께 사용하기 위한 경량 데이터베이스 마이그레이션 도구입니다. Django에서는 자체적으로 DB의 schema 구조 변경에 따라 Migration file을 생성해서 적용하거나 이전 기록을 추적해서 rollback하는 기능이 있습니다. 하지만 Django 없이 Flask나 Fastapi을 통해 Web backend를 개발하는 환경에서는 이런 DB관련해서 지원하지 않기 때문에 SQLAlchemy를 설치해서 DB Table을 설계하고 Alembic으로 Migration, rollback을 진행하게 됩니다.

 

설명은 이정도로 충분한 것 같습니다. 이제 Alembic Tutorial을 시작해 보겠습니다. Alembic Tutorial을 진행하려면 사전에 fastapi, SQLAlchemy가 동작하는 환경이 있어햐 합니다. 간단한 환경구성과 예제 코드를 작성해 보도록 하겠습니다.

 

 

---

환경 구성

fastapi, uvicorn, sqlalchemy, Alembic를 설치해 보도록 하겠습니다.

pip install fastapi uvicorn SQLAlchemy alembic python-dotenv

 

 

저는 추가적으로 postgresql을 사용하기 때문에 python과 Database를 이어줄 adapter인 psycopg2-binary를 설치하겠습니다. 다른 DB를 사용하시는 분들은 그에 맞는 adapter를 설치해주시기 바랍니다.

pip install psycopg2-binary

 

 

연결된 localDB를 확인하기 위해 Dbeaver를 설치해 주도록 합시다.

DBeaver Community | Free Universal Database Tool

 

 

그리고 Local DB는 본인의 상황에 맞게 생성해 주도록 합니다. 저는 DB container를 docker compose를 통해 생성하겠습니다.

(Alembic Tutorial이 필요하신 분이라면 DB는 있을 것으로 예상되어 DB 구축을 자세하게 작성하지는 않겠습니다)

services:
  local_backend:
    image: python:3.12.6-slim
    container_name: local_backend
    volumes:
      - .:/app
    working_dir: /app
    ports:
      - "${BACKEND_PORT}:${BACKEND_PORT}"
    depends_on:
      - local_db
      - server_db
    command: tail -f /dev/null

  local_db:
    image: ${DB_IMAGE}
    container_name: ${LOCAL_DB_CONTAINER_NAME}
    environment:
      POSTGRES_USER: ${LOCAL_DB_USER}
      POSTGRES_PASSWORD: ${LOCAL_DB_PASSWORD}
      POSTGRES_DB: ${LOCAL_DB_NAME}
    ports:
      - "${LOCAL_DB_PORT}:${DB_DEFAULT_PORT}"

  server_db:
    image: ${DB_IMAGE}
    container_name: ${SERVER_DB_CONTAINER_NAME}
    environment:
      POSTGRES_USER: ${SERVER_DB_USER}
      POSTGRES_PASSWORD: ${SERVER_DB_PASSWORD}
      POSTGRES_DB: ${SERVER_DB_NAME}
    ports:
      - "${SERVER_DB_PORT}:${DB_DEFAULT_PORT}"

 

 

docker compose로 python을 실행시킬 container, SQLAlchemy로 table을 개발에 따라 자유롭게 다룰 local_db를 host에서 5432 port, 개발이 끝나고 검증이 필요한 server_db container를 5433 port로 접근할 수 있도록 만들었습니다.

 

 

.env 파일은 아래와 같이 작성했습니다. password는 알아서 입력해 주세요.

(당연한 얘기겠지만 Server Environment 변수는 원래는 여기에 기입하지 않고 server에 migration을 할 instance에 따로 기입 되어 있어야 합니다. 이 글은 예제이니 Server변수가 여기에 작성이 된 것입니다.)

## DB ENVIRONMENT
DB_TYPE=postgresql
DB_IMAGE=postgres:16.4-bookworm
DB_DEFAULT_PORT=5432


## SERVER DB ENVIRONMENT
LOCAL_DB_HOST=local_db

LOCAL_DB_CONTAINER_NAME=local_db
LOCAL_DB_USER=tyoon9781
LOCAL_DB_PASSWORD=...
LOCAL_DB_NAME=localdb
LOCAL_DB_PORT=5432


## SERVER DB ENVIRONMENT
SERVER_DB_HOST=server_db

SERVER_DB_CONTAINER_NAME=server_db
SERVER_DB_USER=tyoon9781
SERVER_DB_PASSWORD=...
SERVER_DB_NAME=serverdb
SERVER_DB_PORT=5433

 

 

DB 생성이 완료되었으면 DBeaver로 접속을 확인해 봅니다. host에서 localdb, serverdb가 접속이 잘 되는 것을 확인할 수 있습니다.

port별로 잘 접속이 된다.

 

 

 

---

예제 코드 작성

예제 코드를 간단하게 작성해 보겠습니다. 

 

[local_main.py]

## [INIT] GET ENV
from dotenv import load_dotenv
load_dotenv()


## [INIT] LOCAL DB URL
from app.db.config import DBConnection
DBConnection.set_db_url(is_server=False)


## [APP] FastAPI
from fastapi import FastAPI
from app.db.connection import create_tables, drop_tables
from app import api

app = FastAPI(on_startup=[create_tables],on_shutdown=[drop_tables])
app.include_router(api.router)


if __name__ == "__main__":
    import uvicorn
    uvicorn.run("local_main:app", host="0.0.0.0", port=8881, reload=True)

 

 

[server_main.py]

## [INIT] GET ENV
from dotenv import load_dotenv
load_dotenv()


## [INIT] LOCAL DB URL
from app.db.config import DBConnection
DBConnection.set_db_url(is_server=True)


## [APP] FastAPI
from fastapi import FastAPI
from app import api


app = FastAPI()
app.include_router(api.router)


if __name__ == "__main__":
    import uvicorn
    uvicorn.run("server_main:app", host="0.0.0.0", port=8882, reload=True)

 

 

[app/api.py]

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

from app.db.connection import get_db
from app.db.schema import *
from app.db.models import *


router = APIRouter()


## API
@router.post("/items/", response_model=ItemRead)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = ItemModel(**item.model_dump())
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item


@router.get("/items/{item_id}", response_model=ItemRead)
def read_item(item_id: int, db: Session = Depends(get_db)):
    db_item = db.query(ItemModel).filter(ItemModel.id == item_id).first()
    if not db_item:
        raise HTTPException(status_code=404, detail="No Item")
    return db_item

 

 

[app/db/config.py]

import os

class DBConnection:
    db_url:str = None

    @classmethod
    def set_db_url(cls, is_server:bool=False):
        db_type = os.getenv("DB_TYPE")
        db_port = os.getenv("DB_DEFAULT_PORT")

        if is_server:
            db_user = os.getenv("SERVER_DB_USER")
            db_pwd  = os.getenv("SERVER_DB_PASSWORD")
            db_host = os.getenv("SERVER_DB_HOST")
            db_name = os.getenv("SERVER_DB_NAME")
        else:
            db_user = os.getenv("LOCAL_DB_USER")
            db_pwd  = os.getenv("LOCAL_DB_PASSWORD")
            db_host = os.getenv("LOCAL_DB_HOST")
            db_name = os.getenv("LOCAL_DB_NAME")
            
        cls.db_url = f"{db_type}://{db_user}:{db_pwd}@{db_host}:{db_port}/{db_name}"

 

 

[app/db/connection.py]

from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker
from app.db.config import DBConnection


Base = declarative_base()
engine = create_engine(DBConnection.db_url)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


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


def create_tables():
    Base.metadata.create_all(bind=engine)


def drop_tables():
    Base.metadata.drop_all(bind=engine)

 

 

[app/db/models.py]

from sqlalchemy import Column, Integer, String, DateTime
from app.db.connection import Base
from datetime import datetime, timezone


## DB Table Schema
def utc_now():
    return datetime.now(tz=timezone.utc)


class ItemModel(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, nullable=True)
    created_at = Column(DateTime(timezone=True), default=utc_now)
    updated_at = Column(DateTime(timezone=True), default=utc_now, onupdate=utc_now)

 

 

[app/db/schemas.py]

from pydantic import BaseModel
from datetime import datetime


class ItemCreate(BaseModel):
    name: str
    description: str|None = None


class ItemUpdate(BaseModel):
    description: str|None = None


class ItemRead(BaseModel):
    id: int
    name: str
    description : str|None = None
    created_at: datetime
    updated_at: datetime
    
    class Config:
        from_attributes = True

 

이렇게 작성하면 local_main.py와 server_main.py는 같은 application 동작을 하게 되지만 DB URL을 서로 다르게 하여 동작하게 되고, 또 startup, shutdown 동작에서 local_main.py는 항상 table을 새로 생성, 삭제하게 됩니다. local_main.py에서 test를 진행해보겠습니다.

 

 

---

1. local_main.py

local_main.py에서는 fastapi startup에 table 생성 구문이 있었으므로 local_main.py를 시작하면 DB table이 생성될 것입니다. 먼저 fastapi를 구동합니다.

 

fastapi 구동이 성공했음을 확인했으면 이제 DBeaver로 DB table을 확인해 보도록 하겠습니다. item table이 생긴 것을 확인할 수 있습니다.

 

 

기본 동작 Test도 해보겠습니다. curl로 직접 보내거나 postman으로 post request를 보낼 수 있습니다.

Postman API Platform | Sign Up for Free

 

 

postman 설치가 귀찮으신 분들은 아래의 curl request를 console, cmd, powershell 등에서 시도해 보시기 바랍니다.

curl -X POST http://127.0.0.1:8881/items/ \
     -H "Content-Type: application/json" \
     -d '{"name": "test_name", "description": "test_description"}'

 

 

Postman으로 Create Item API를 request하면 item 생성이 정상적으로 됩니다.

 

 

DB도 확인해보면 item이 제대로 들어간 것을 확인할 수 있습니다.

 

 

Item 조회도 잘 되는 것을 확인할 수 있습니다. 1번 item에 대한 정보를 확인할 수 있습니다.

 

이제 local_main.py로 실행한 fastapi를 종료하겠습니다. ctrl+c를 누르면 종료됩니다. 그리고 종료됨과 동시에 DB의 table이 삭제된 것을 확인할 수 있습니다.

 

 

이번에는 server_main.py를 동작해보도록 하겠습니다.

 

 

---

2. server_main.py

server_main은 동작을 시켜도 table을 만드는 구문이 없으니 DB에 table이 생성되지 않습니다. 우선 server_main.py를 실행해 보겠습니다.

 

 

그리고 DBeaver에서 serverdb를 확인해보면 table이 없는 것을 확인할 수 있습니다.

 

 

이렇게 되면 api는 당연히 동작하지 않기 때문에 여기서 server_main.py를 종료하겠습니다. 이제 기능 확인은 다 되었으니 Alembic을 사용하기 위해 간단한 준비를 하겠습니다.

 

 

---

Init Alembic 

먼저 alembic을 시작합니다. alembic init alembic을 입력합니다. 

alembic init alembic

 

그러면 alembic.ini 파일과 alembic/ 폴더가 생깁니다. alembic init [keyword]로 폴더명이 정해지는 것인데 alembic.ini의 파일 이름은 바뀌지 않습니다. alembic init alembic이 가장 관리하기 편할 것입니다.

 

먼저 alembic.ini 파일의 sqlalchemy.url 항목을 찾습니다. 그리고 비워줍시다. 여기에 값을 넣기에는 .env로 작성된 환경변수는 적절하지 않습니다. 

 

sqlalchemy.url =

 

 

그 다음에는 alembic/env.py에 들어가서 sqlalchemy.url에 값을 넣어주도록 하겠습니다. 이렇게 하면 .env 기반으로 sqlalchemy.url에 db_url을 넣을 수 있게 됩니다.

from logging.config import fileConfig

from sqlalchemy import engine_from_config
from sqlalchemy import pool

from alembic import context

from dotenv import load_dotenv
load_dotenv()

from app.db.config import DBConnection
DBConnection.set_db_url(is_server=True)

# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config
config.set_main_option("sqlalchemy.url", DBConnection.db_url)

 

 

그 다음에는 target_metadata 항목을 찾아봅니다. 찾았으면 아래와 같이 입력합니다.

from app.db.models import *
from app.db.connection import Base
target_metadata = Base.metadata

 

 

이제 모든 사전작업이 끝났습니다! alembic에서 해볼 동작은 다음과 같습니다.

1. First Migration
2. table 추가, 변경 후 migration
3. Migration 이전 Version으로 Rollback

먼저 First Migration부터 해보겠습니다.

 

 

---

1. First Migration

아래와 같이 입력합니다. 현재의 Table 구조를 commit한다고 생각하면 됩니다. 이 명령어 만으로는 DB에 적용되지는 않습니다.

(--autogenerate는 upgrade, downgrade 함수의 내용을 자동으로 작성해주는 옵션입니다)

alembic revision --autogenerate -m "first migration"

 

 

그러면 alembic/versions/ 폴더에 xxxx_first_migration.py가 생긴 것을 확인할 수 있습니다.

 

 

xxxx_first_migration.py 파일을 열어보면 upgrade, downgrade 함수가 있는 것을 확인할 수 있습니다. 이 migration file을 DB에 적용하면 upgrade 함수가 적용되는 것이고 migration 된 것을 rollback를 하게 되면 downgrade함수가 동작합니다.

우리는 이 upgrade함수를 DB에 적용하기 위해 아래와 같이 입력할 것입니다.

alembic upgrade head

 

transactional DDL이 적용되었다.

 

 

 

alembic upgrade head를 입력하게 되면 migration 된 파일들의 가장 최신본까지 전부 반영하게 됩니다. 반영하게 되면 server_db에는 table이 생성되었을 것입니다. 확인해 보겠습니다.

 

 

위 그림을 보면 alembic_version table과 items table이 생긴 것을 확인할 수 있습니다. alembic_version table은 migration의 현재 version을 확인할 수 있는 table입니다.

(안타깝게도 alembic/versions/ 에 생성되는 파일명만으로는 전체 순서를 알기 어렵습니다. 파일 내 revision, down_revision을 하나하나 보며 추적하거나 생성 시점같은 것으로 유추하는 것이 전부입니다)

 

여기까지 First Migration이었습니다. 이제 table을 변경해 보도록 하겠습니다.

 

 

---

2. table 추가, 변경 후 migration

table을 변경, 추가해 보도록 하겠습니다. 먼저 item에 data를 추가하겠습니다.

 

 

그 다음에 table의 column 하나를 삭제해보도록 하겠습니다.

[app/db/models.py]

class ItemModel(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    # description = Column(String, nullable=True)
    created_at = Column(DateTime(timezone=True), default=utc_now)
    updated_at = Column(DateTime(timezone=True), default=utc_now, onupdate=utc_now)

 

 

이번에는 새로운 table을 만들어 보겠습니다.

[app/db/models.py]

class User(Base):
    __tablename__ = "user"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True)
    is_admin = Column(Boolean, index=True)
    created_at = Column(DateTime(timezone=True), default=utc_now)
    updated_at = Column(DateTime(timezone=True), default=utc_now, onupdate=utc_now)

 

 

이 상태로 migration을 진행하면 어떻게 될까요? 아래와 같이 입력해 보겠습니다.

alembic revision --autogenerate -m "[Item] delete description column, [User] create table"

 

 

 

 

migration 파일이 생성되었습니다. alembic/versions/ 폴더에 가면 새로운 파일이 생성되어 있습니다.

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('user',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('name', sa.String(), nullable=True),
    sa.Column('email', sa.String(), nullable=True),
    sa.Column('is_admin', sa.Boolean(), nullable=True),
    sa.Column('created_at', sa.DateTime(timezone=True), nullable=True),
    sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
    sa.PrimaryKeyConstraint('id'),
    sa.UniqueConstraint('email')
    )
    op.create_index(op.f('ix_user_id'), 'user', ['id'], unique=False)
    op.create_index(op.f('ix_user_is_admin'), 'user', ['is_admin'], unique=False)
    op.create_index(op.f('ix_user_name'), 'user', ['name'], unique=False)
    op.drop_column('items', 'description')
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('items', sa.Column('description', sa.VARCHAR(), autoincrement=False, nullable=True))
    op.drop_index(op.f('ix_user_name'), table_name='user')
    op.drop_index(op.f('ix_user_is_admin'), table_name='user')
    op.drop_index(op.f('ix_user_id'), table_name='user')
    op.drop_table('user')
    # ### end Alembic commands ###

 

 

그런데 이대로 진행하면 description Column이 삭제되면서 data도 삭제가 될텐데 DB에 적용이 가능할까요?

alembic upgrade head

 

 

놀랍게도 오류 없이 migration을 완료했습니다.

 

 

그리고 table에는 Item의 description column이 삭제된 것을 알 수 있습니다. User Table도 생성이 되었습니다.

 

 

이로써, table의 변경, 생성이 제대로 적용된 것을 확인할 수 있었습니다. 이 tutorial을 통해 alembic으로 migration을 할 때는 data가 있는 column 삭제에 대해서 미리 감지를 해주지 않는 다는 것을 알 수 있습니다. alembic으로 column을 삭제하려는 시도를 할 때는 data가 삭제되어도 괜찮은지를 제대로 확인하고 진행을 해야 합니다. 그렇지 않으면 snapshot을 뒤져가며 데이터 원복을 위해 눈물나는 야근을 할지도 모릅니다....

 

이번에는 이전 version으로 rollback을 해보도록 하겠습니다.

 

 

 

---

3. Migration 이전 Version으로 Rollback

rollback은 어떻게 하면 좋을까요? 이전에 되었던 versions/에 있는 migration file중에 원하는 것을 찾아 선택하면 됩니다. 저는 여기서 "first migration"을 선택하겠습니다.

 

alembic에서 upgrade, downgrade를 하는 방식은 2가지 방법이 있습니다. 하나는 단계(숫자)를 입력하는 것이고 다른 하나는 version을 입력하는 것입니다. 제 "first migration"의 Revision ID는 8ac44250a772입니다. 그리고 이 revision은 한 단계 아래입니다. 그러므로 다음과 같이 작성할 수 있습니다.

## version으로 downgrade
alembic downgrade 8ac44250a772

## 단계로 downgrade
alembic downgrade -1

 

 

downgrade를 적용해 보도록 하겠습니다.

 

 

적용한 결과 DB에 User table은 삭제되었고 description column은 돌아왔습니다. 하지만 description에 data는 없습니다.

(만약 description column에 nullable = False로 설정했었다면 에러가 발생합니다. 에러가 발생하지 않게 하려면 downgrade 함수에 server_default='dummy_data'같은 방식으로 데이터를 넣는 것으로 조치 할 수 있습니다.)

 

 

여기까지 alembic tutorial이었습니다 감사합니다.

 

 

---

Outro

alembic을 잘 활용하면 개발자가 DB Table을 revision 단계별로 적용할 수 있게 됩니다. 물론 DBA가 직접 하는 것만큼 최적화를 하기는 어렵습니다. 하지만 alembic과 SQLAlchemy을 통해 DB의 migration을 보다 잘 관리할 수 있는 것도 사실입니다. alembic을 잘 사용한다면 version 관리, 자동 migration, script 재사용, python으로 migration이 가능, customizing같은 장점을 극대화 시켜 데이터베이스 스키마를  잘 관리할 수 있을 거라 생각합니다. 감사합니다.

 

 

---

 

*reference

https://alembic.sqlalchemy.org/en/latest/tutorial.html

https://www.postman.com/

https://dbeaver.io/