[Redis] - Tutorial with Python

 

*목차

 - Intro

 - 환경설정

 - 예제코드(Redis 기능 제외)

 - Redis 없이 기능 Test

 - Redis 적용

 - Redis 적용 후 기능 Test

 - Outro

 

---

Intro

안녕하세요, 이번에는 Redis를 Python에서 활용하는 방법에 대해서 알아 보겠습니다. Redis 자체는 Python만을 위한 것이 아닌 In-memory 방식의 DBMS입니다. 그래서 category도 DB에 넣을까 하다가 Python에서 사용하는 Redis에 대한 문법이 주로 소개될 것 같아 Python Library로 하여 소개 드립니다.

 

Redis는 다음과 같은 특징을 가지고 있습니다.

• In-memory : Redis는 data를 기본적으로 disk가 아닌 memory에 저장합니다.
  • 저장 관점 : MySQL, PostgreSQL같은 DBMS는 기본적으로 Disk에 data를 저장합니다. 그래서 Process가 정상적으로 종료된다면 지금까지 기록한 data는 휘발되지 않고 disk에 남아있게 됩니다. 하지만 redis는 Process를 종료하게 되면 data가 휘발 됩니다. 만약 데이터 휘발을 방지하고 싶다면 Persistence Option을 사용해도 되지만 여기서는 다루지 않겠습니다.
  • 속도 관점 : disk보다는 memory에서 불러오는 데이터가 아무래도 속도가 더 빠르겠죠? 수많은 요청이 들어왔을 때 memory에서 읽어서 보내주는 방식과 disk에서 query를 실행해서 data를 읽어오는 방식은 resource 사용 측면에서 많은 차이가 있습니다. 
    (물론 MySQL, PostgreSQL같은 DBMS도 자체적으로 cache가 있어서 중복된 요청에 대해서 빠르게 응답할 수 있도록 지원합니다.) 
• NoSQL : 전통적으로 쓰이는 RDBMS가 아닌 key-value  기반의 데이터 저장 방식입니다.
  • key : 고유한 식별자로 데이터를 찾기 위한 인덱스 역할을 합니다. key 값으로 String을 사용할 수 있습니다.
  • Value : key에 1:1로 대응되는 데이터입니다. 단순한 문자열 외에도 다양한 자료형을 저장할 수 있습니다.
• 자료형 : Redis에서는 아래와 같은 자료형을 지원합니다. 
  • String: 가장 기본적인 자료형이며, 단순 문자열 또는 숫자를 저장합니다.
    • 예시: SET "user:123" "Alice"
    • 읽기: GET "user:123"
  • Hash: 필드와 값을 포함하는 해시맵입니다. 데이터베이스의 행(row)처럼 특정 객체의 여러 속성을 저장할 수 있습니다.
    • 예시: HSET "user:123" "name" "Alice" "age" "30"
    • 읽기: HGET "user:123" "name"
  • List: 삽입 순서가 유지되는 리스트입니다. 대기열이나 스택으로 사용할 수 있습니다.
    • 예시: LPUSH "queue" "task1" (리스트 앞에 추가), RPUSH "queue" "task2" (리스트 뒤에 추가)
    • 읽기: LPOP "queue" (리스트 앞에서 읽고 삭제)
  • Set: 중복되지 않는 값들의 집합입니다. 교집합, 합집합 등의 집합 연산이 가능합니다.
    • 예시: SADD "tags" "python" "fastapi"
    • 읽기: SMEMBERS "tags"
  • Sorted Set: 값에 점수를 매겨 정렬된 집합을 저장합니다.
    • 예시: ZADD "leaderboard" 100 "user1" 200 "user2"
    • 읽기: ZRANGE "leaderboard" 0 -1 WITHSCORES

 

이런 특징이 있는 Redis는 주로 Cache, Session 관리, 실시간 데이터 처리에 활용됩니다. Cache 예제를 통해 redis의 간단한 사용법을 알아봅시다.

 

 

---

환경 설정

이 글에서 DB 조회는 DBeaver를 사용합니다.

DBeaver Community | Free Universal Database Tool

 

이 글에서 API Test는 Postman을 사용합니다.

Postman API Platform | Sign Up for Free

 

이 글에서 Container는 Docker Compose를 통해 관리합니다.사용합니다. docker compose로 fastapi, postgresql, redis container를 각각 만들어 보도록 하겠습니다.

 

docker-compose.yml 파일입니다.

services:
  backend:
    image: ${BACKEND_IMAGE}
    container_name: backend
    volumes:
      - .:${BACKEND_DIR}
    working_dir: ${BACKEND_DIR}
    ports:
      - "${BACKEND_PORT}:${BACKEND_PORT}"
    depends_on:
      - db
      - redis
    command: tail -f /dev/null

  db:
    image: ${DB_IMAGE}
    container_name: ${DB_CONTAINER_NAME}
    environment:
      POSTGRES_USER: ${DB_USER}
      POSTGRES_PASSWORD: ${DB_PASSWORD}
      POSTGRES_DB: ${DB_NAME}
    volumes:
      - ./postgresql_data:/var/lib/postgresql/data
    ports:
      - "${DB_PORT}:${DB_PORT}"

  redis:
    image: ${REDIS_IMAGE}
    container_name: ${REDIS_CONTAINER_NAME}
    ports:
      - "${REDIS_PORT}:${REDIS_PORT}"

 

 

yml 파일을 잘 보시면 volumes가 있습니다. 이 yml 실행 위치에 ./postgresql_data 폴더를 미리 생성하셔서 postgresql의 data를 disk에 저장할 수 있도록 합니다.

 

 

.env는 다음과 같습니다. password 부분은 각자 알아서 기입해주시기 바랍니다.

## BACKEND
BACKEND_IMAGE=python:3.12.6-slim
BACKEND_PORT=8000
BACKEND_DIR=/app

## DB
DB_TYPE=postgresql
DB_IMAGE=postgres:16.4-alpine
DB_HOST=db
DB_CONTAINER_NAME=mydb
DB_USER=tyoon9781
DB_PASSWORD=...
DB_NAME=localdb
DB_PORT=5432

## REDIS
REDIS_IMAGE=redis:7.4-alpine
REDIS_CONTAINER_NAME=myredis
REDIS_USER=tyoon9781
REDIS_PASSWORD=...
REDIS_PORT=6379

 

 

Python Package는 아래와 같습니다. requirements.txt과 dockerfile을 사용하셔도 좋고 container 안에서 직접 설치하셔도 됩니다. 저는 container 안에서 직접 설치했습니다.

pip install fastapi uvicorn SQLAlchemy python-dotenv psycopg2-binary redis

 

 

 

---

예제코드 (Redis 기능 제외)

Redis 기능이 제외된 예제코드를 작성해 보겠습니다.

 

 

[main.py]

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


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

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


if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app", host="0.0.0.0", port=int(os.getenv('BACKEND_PORT')), 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.schemas 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


@router.put("/items/{item_id}", response_model=ItemRead)
def update_item(item_id: int, item: ItemUpdate, 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 found")
    
    # Update the fields in the db_item with values from the item
    for key, value in item.model_dump().items():
        setattr(db_item, key, value)
    
    db.commit()
    db.refresh(db_item)
    return db_item


@router.delete("/items/{item_id}", response_model=dict)
def delete_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 found")
    
    db.delete(db_item)
    db.commit()
    return {"detail": "Item deleted successfully"}

 

 

[app/db/connection.py]

from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker
import os


db_type = os.getenv("DB_TYPE")
db_port = os.getenv("DB_PORT")
db_user = os.getenv("DB_USER")
db_pwd  = os.getenv("DB_PASSWORD")
db_host = os.getenv("DB_HOST")
db_name = os.getenv("DB_NAME")
db_url = f"{db_type}://{db_user}:{db_pwd}@{db_host}:{db_port}/{db_name}"

Base = declarative_base()
engine = create_engine(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

 

 

이 상태로 동작하면 Redis container와는 통신하지 않고 postgresql container와 직접 통신하면서 Item Data를 CRUD하게 됩니다. DBeaver를 통해 확인해 보니 Data가 잘 생성된 것을 확인할 수 있습니다.

 

그리고 3번 데이터를 조회 해보겠습니다.

 

잘 조회가 되는 것을 확인할 수 있습니다. 이 동작은 얼마나 걸렸을까요? Postman으로 test해보도록 하겠습니다.

 

 

 

---

Redis 없이 기능 Test

fastapi에서 postgresql에 data를 100번 요청해 보도록 하겠습니다. 그리고 평균값을 구해보겠습니다. Test Tool은 Postman의 Run Collection을 활용했습니다.

get 요청을 100번 시도한다.

 

 

결과를 확인해 보겠습니다. 100번 시도한 결과 평균 9ms (CPU : Intel i7-12700k) 정도 발생했습니다. 

 

물론 data의 크기도 작고, postgresql의 자체적으로 cache를 가지고 있기 때문에 반복적인 요청에 대해서 자체적으로 최적화를 하기 때문에 꽤 낮은 수치가 나왔습니다.

 

여기서 redis를 적용하면 어떻게 될까요? 과연 이정도 작은 작업에도 속도를 줄여주는데 도움이 될 수 있을까요?

 

 

 

---

Redis 적용

redis를 사용하기 위해 code에 몇가지 내용을 추가합니다.

 

[app/db/connections.py]

...

## redis
from redis import Redis
redis_client = Redis(host="redis", port=int(os.getenv("REDIS_PORT")), db=0)
redis_ttl = 60  ## 1 minute

...

 

 

[app/api.py]

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

from app.db.connection import get_db, redis_client, redis_ttl
from app.db.schemas import *
from app.db.models import *
import json


router = APIRouter()


def cache_item(db_item:ItemModel):
    ## redis
    item_str = ItemRead.model_validate(db_item).model_dump_json()
    redis_client.set(f"item:{db_item.id}", item_str, ex=redis_ttl)


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

    ## redis
    cache_item(db_item)

    return db_item


@router.get("/items/{item_id}", response_model=ItemRead)
def read_item(item_id: int, db: Session = Depends(get_db)):
    ## redis
    cached_item = redis_client.get(f"item:{item_id}")

    if cached_item:
        return ItemRead.model_validate(json.loads(cached_item))

    db_item = db.query(ItemModel).filter(ItemModel.id == item_id).first()
    if not db_item:
        raise HTTPException(status_code=404, detail="No Item")
    
    ## redis
    cache_item(db_item)

    return db_item


...

 

 

이렇게 하면 post를 할 때는 기존보다 작업이 추가되긴 했지만 get을 할 때는 redis에 먼저 data가 있는지 확인 후 데이터가 있으면 그대로 return하는 코드가 생겼습니다. 한 번 test 해보겠습니다.

 

 

 

---

Redis 적용 후 기능 Test

 

테스트 결과 7ms (CPU : Intel i7-12700k) 정도 나옵니다. 차이가 정말 안나지만 그래도 좀 단축되기는 했습니다.

 

 

 

---

Outro

사실 이 글을 작성할 때도 redis의 활용을 어떻게 하는지만 작성하기를 원했고 실제 성능의 향상에는 큰 기대를 하지 않았습니다. 이유는 아래와 같습니다.

1. DB에 유의미한 부하를 주는 Code를 작성하기에는 Tutorial 수준으로 작성하기는 어렵다
2. Postgresql도 자체적으로 반복적인 요청을 빠르게 처리하기 위한 cache가 있다. (최소 128MB) 이 cache를 이용하지 않는 요청을 해야 disk vs memory 방식을 볼 수 있을텐데 Tutorial 수준에서는 작성하기 어렵다.

다행인(?) 점은 둘 다 속도가 매우 비슷하고 redis가 약간 더 빠르게 나와서 redis가 부정적으로 보이지는 않는다는 점 입니다...

 

Redis를 활용해서 Item을 cache에 저장 후 조회 시도를 해보았습니다. 아쉽게도 큰 성능차이를 확인할 수는 없었지만 나중에 정말로 많은 사용자들로 인해 DB에 부하가 생기는 시점에는 Redis가 정말 좋은 Solution이 될 것으로 생각합니다. 감사합니다.

 

 

 

---

 

*reference

https://www.postman.com/

https://dbeaver.io/

https://redis.io/