[FastAPI] FastAPI에서 Redis 사용하기

1. Redis 개념

1.1. Redis란 무엇인가?

• Redis는 고성능의 오픈 소스 인메모리 데이터베이스로, 키-값 데이터 구조를 지원.
• 데이터가 메모리에 저장되어 매우 빠른 읽기/쓰기 속도를 제공
• 단순한 문자열뿐만 아니라 다음과 같은 데이터 구조를 지원
  • Lists: 순서가 있는 값의 컬렉션 (예: 대기열, 스택)
  • Sets: 중복 없는 값의 집합
  • Sorted Sets: 우선순위를 가진 값의 집합
  • Hashes: 필드와 값을 가진 객체
  • Streams: 로그 데이터와 실시간 메시지 스트림
  • Bitmaps 및 HyperLogLogs: 특수한 데이터 처리 용도
• 특정 키에 TTL(Time To Live)을 설정해 자동으로 데이터를 만료시킬 수 있다.

1.2. 언제 사용하는가?

• 캐싱 (예: 데이터베이스 질의 결과 캐싱)
• 세션 저장소 (예: 사용자 로그인 세션 관리)
• 실시간 분석 (예: 실시간 사용자 수 추적)
• 메시지 큐 (Pub/Sub 모델)
• Queue 관리

---

2. Redis와 Python 연동

2.1. 설치

pip install redis (동기 처리용)
pip install aioredis (비동기 처리용)

 

2.2. 기본 사용법 

• redis 클라이언트 생성

import redis

redis_client = redis.Redis(host='localhost', port=6379, db=0)

 

• Set: key, value, ex(TTL 초 단위)
• Get:name

# 데이터 저장
redis_client.set(name='mykey', value='Hello, Redis!', ex=10)

# 데이터 가져오기
value = redis_client.get('mykey')
print(value.decode())  # Hello, Redis!

 

유용한 기능

• delete: name
• setex
  • 설명: 값을 저장하고 TTL(초 단위)을 설정
  • 파라미터: name, time(초 단위), value 저장할 값
• exists: name
• keys
  • 패턴에 매칭되는 키를 가져옵니다.
  • 파라미터: pattern (str): 매칭할 키 패턴 (*는 모든 키).
• ttl
  • 키의 남은 TTL(초 단위)을 반환
  • 파라미터: name

# 키 삭제
redis_client.delete('mykey')

# 10초 뒤에 만료
redis_client.setex(name='tempkey', time=30, value='Temporary Value')

# 존재 여부 확인
exists = redis_client.exists('mykey')
print(exists)  # 1(존재함), 0(존재하지 않음)

# 모든 키 다져오기
keys = redis_client.keys(pattern='*')
print(keys)  # [b'mykey', b'mylist', b'myhash']

# 키 남은 시간 확인
ttl = redis_client.ttl('tempkey')
print(ttl)  # 남은 시간(초)

 

2.3. 리스트 

• lpush
  • 스트의 왼쪽(앞)에 하나 이상의 값을 추가
  • 파라미터: name(리스트이름), values(추가할 값)
• rpush
  • 리스트의 오른쪽(뒤)에 하나 이상의 값을 추가
• lpop
  • 리스트의 왼쪽(앞)에서 값을 제거하고 반환
  • 파라미터: name
• rpop
  • 리스트의 오른쪽(뒤)에서 값을 제거하고 반환
• lrange
  • 지정한 범위의 리스트 요소를 가져옴
  • 파라미터: name(리스트 이름), start, end(-1은 리스트 끝까지)
• llen
  • 리스트의 길이(요소 개수)를 반환
• lindex
  • 지정한 인덱스의 요소를 반환
  • 파라미터: name, index(가져올 인덱스, 0부터 시작)
• lset
  • 특정 인덱스에 값을 설정
  • 파라미터: name, index, value
• ltrim
  • 리스트를 지정한 범위로 자름(나머지 요소 삭제)
  • 파라미터: name, start, end
• blpop / brpop
  • 리스트의 왼쪽(blpop) 또는 **오른쪽(brpop)**에서 값을 제거하고 반환하며, 값이 없으면 대기
  • 파라미터: keys, timeout(대기시간 초 단위)

# 리스트에 값 추가
redis_client.lpush(name='mylist', *['value1', 'value2'])
redis_client.rpush(name='mylist', *['value3', 'value4'])

redis_client.lpop(name='mylist')
redis_client.rpop(name='mylist')

# 값 범위 조회
redis_client.lrange(name='mylist', start=0, end=-1)

# 값 개수 조회
redis_client.llen(name='mylist')

# 인덱스로 조회
redis_client.lindex(name='mylist', index=1)

# 인덱스로 값 설정
redis_client.lset(name='mylist', index=1, value='new_value')

# 인덱스로 삭제
redis_client.ltrim(name='mylist', start=0, end=1)

# 리스트 값 가져오기
values = redis_client.lrange('mylist', 0, -1)
print([v.decode() for v in values])  # ['C', 'B', 'A']

# 기다렸다가 제거 및 반환
redis_client.blpop(keys=['mylist'], timeout=10)

 

2.4. 해시 (HSET, HGET)

• 해시: 필드와 값을 저장할 수 있는 구조

# 해시에 값 추가
redis_client.hset('myhash', 'field1', 'value1')
redis_client.hset('myhash', 'field2', 'value2')

# 해시에서 특정 필드 가져오기
value = redis_client.hget('myhash', 'field1')
print(value.decode())  # value1

 

2.5. 집합 (SADD, SMEMBERS)

집합은 중복 없는 데이터를 저장합니다.

# 집합에 값 추가
redis_client.sadd('myset', 'A', 'B', 'C')

# 집합의 모든 값 가져오기
values = redis_client.smembers('myset')
print([v.decode() for v in values])  # ['A', 'B', 'C']

 

2.6. 정렬된 집합 (ZADD, ZRANGE)

점수와 함께 데이터를 저장하며 정렬된 순서로 데이터를 가져옵니다.

# 정렬된 집합에 값 추가
redis_client.zadd('myzset', {'A': 1, 'B': 2, 'C': 3})

# 점수 순으로 값 가져오기
values = redis_client.zrange('myzset', 0, -1, withscores=True)
print(values)  # [(b'A', 1.0), (b'B', 2.0), (b'C', 3.0)]

 

2.7. Pipeline 사용

# Pipeline 생성
with redis_client.pipeline() as pipe:
    pipe.set('key1', 'value1')
    pipe.set('key2', 'value2')
    pipe.get('key1')
    result = pipe.execute()

print(result)  # [True, True, b'value1']

 

---

3. FastAPI에서 사용하기

3.1. Depends로 의존성 주입

from fastapi import FastAPI, Depends
import redis

def get_redis():
    return redis.Redis(host="localhost", port=6379, db=0)

app = FastAPI()

@app.get("/cache")
def get_cache(key: str, redis_client: redis.Redis = Depends(get_redis)):
    value = redis_client.get(key)
    return {"key": key, "value": value.decode() if value else None}

3.2. 연결 관리

Redis는 네트워크 연결을 사용하기 때문에 연결 관리가 중요합니다. FastAPI에서는 앱 시작 시 연결을 초기화하고 앱 종료 시 연결을 종료하도록 설정해 준다.

redis_client = redis.Redis(host='localhost', port=6379, db=0)

redis_client.close()

3.3. 비동기 사용

from fastapi import FastAPI, Depends
import aioredis

app = FastAPI()
redis_client = None

async def get_redis_client():
    global redis_client
    if not redis_client:
        redis_client = await aioredis.from_url("redis://localhost")
    return redis_client

@app.get("/async-cache")
async def get_cache(key: str, redis_client = Depends(get_redis_client)):
    value = await redis_client.get(key)
    return {"key": key, "value": value.decode() if value else None}

@app.post("/async-cache")
async def set_cache(key: str, value: str, redis_client = Depends(get_redis_client)):
    await redis_client.set(key, value)
    return {"message": "Cache saved!"}