Redis란?
Redis는 디스크가 아닌 메인 메모리(RAM)에 데이터를 저장하고 관리하는 In-Memory NoSQL 데이터베이스입니다. 모든 데이터를 메모리에서 처리하기 때문에 디스크 I/O가 발생하는 전통적인 RDBMS에 비해 압도적인 처리 속도를 제공합니다.
| 구분 | RDBMS (예: Oracle, PostgreSQL) | Redis (In-Memory) |
| 저장 매체 | 물리적 디스크 (HDD, SSD) | 메인 메모리 (RAM) |
| 데이터 구조 | 테이블, 행, 열 (정형 데이터) | Key-Value (NoSQL) |
| 응답 속도 | 밀리초(ms) 단위 | 마이크로초(μs) 단위 |
| 영속성 | ACID 원칙에 따른 완전한 보존 | 메모리 휘발성 (Snapshot/AOF 기능 제공) |
| 주요 용도 | 데이터 원본 저장소, 복잡한 쿼리 처리 | 캐싱, 세션 관리, 실시간 랭킹 |
Redis의 주요 용도로 데이터 캐싱이 있습니다.
조회 빈도가 높고 데이터의 변동이 적은 데이터(예: 병원 마스터 리스트)는 캐싱을 통해 시스템 성능을 비약적으로 향상시킬 수 있습니다.
캐싱을 사용할 때 Redis가 사용하는 대표적인 전략은 Cache Aside입니다. 애플리케이션이 Redis와 DB를 독립적으로 관리하며, Redis를 우선 확인하는 패턴입니다.
- 최초 요청 시 Oracle DB에서 HOSPITAL_MASTER 중 미가입 목록을 조회하여 Redis에 unregistered_hospitals 키로 저장합니다.
- 다음으로 사용자가 로그인 할 때 서버가 먼저 Redis에 unregistered_hospitals라는 키가 있는지 확인합니다.
- 있다면(Cache Hit): DB를 건드리지 않고 Redis에 있는 리스트를 즉시 반환합니다.
- 없다면(Cache Miss): 딱 한 번 Oracle DB에서 쿼리를 실행해 데이터를 가져온 뒤, 이를 Redis에 저장하고 사용자에게 줍니다. 이후부터는 다른 사용자들이 올 때 Redis에서 바로 나갑니다.
-
@router.post("") 회원가입 처리 함수가 성공적으로 완료되어 connection.commit( )이 일어나는 시점에, Redis에 저장된 unregistered_hospitals키를 삭제합니다.
Redis 기본 문법
Key-Value (이름표와 내용물)
Redis의 가장 기본은 모든 데이터를 이름표(Key)와 내용물(Value)의 쌍으로 관리한다는 점입니다. 데이터베이스처럼 복잡한 표(Table)를 만들지 않고, 단 두 개의 칸만 있습니다.
SET & GET (넣기와 꺼내기)
Redis의 데이터를 입력하고 내어주는 가장 기본적인 동작입니다.
EX (데이터의 유통기한)
Redis에 저장되는 모든 데이터에는 EX (Expire)라는 유통기한 딱지를 붙일 수 있습니다. EX=60이라고 설정하면, 이 데이터는 정확히 60초 뒤에 스스로 사라집니다.
Cache Hit & Miss (성공과 실패의 기록)
Redis에 데이터를 찾으러 갔을 때 발생하는 두 가지 상황을 부르는 용어입니다.
- Cache Hit (성공): 찾으려는 이름표가 Redis에 있어서 즉시 데이터를 받아갑니다.
- Cache Miss (실패): 이름표가 없어서 빈손으로 돌아와, 메인 DB까지 직접 다녀와야 하는 상황입니다.
DEL (데이터 파기)
DEL 명령어를 쓰면 해당 이름표를 가진 사물함을 즉시 깨끗하게 비워버립니다.
Redis 리소스 생성
Azure 포털에 접속해 작업 중인 리소스 그룹을 선택하고 Marketplace에서 Azure Managed Redis를 선택합니다. (Azure Cache for Redis는 구버전이라고 합니다.)
해당 리소스의 정보를 입력해줍니다.
특히 Data tier에서 In-memory는 모든 데이터를 100% RAM(메모리)에만 저장하는 방식이고 Flash는 자주 쓰는 데이터는 RAM에, 잘 안 쓰는 데이터는 빠른 SSD(NVMe)에 나누어 저장하는 방식입니다.
프로젝트 구성의 편의성을 위해 Network를 열어줍니다.
리소스가 생성되었다면 Redis 사용을 위한 엔드포인트를 확인합니다.
여기서 주의할 점은 기존의 일반적인 "Azure Cache for Redis"는 6380 포트를 사용하지만, 생성한 최신 서비스인 Azure Managed Redis(AMR)는 10000번 포트를 TLS(보안) 접속용 기본 포트로 사용합니다.
또한 접근을 위한 Access keys 옆의 복사 버튼을 누릅니다. 이것이 파이썬 코드에서 사용할 비밀번호입니다.
이제 vscode로 접속하여 pip install redis 명령어를 사용해 python 환경에 redis를 설치합니다.
그리고 아래의 코드를 사용하여 Redis에 정상적으로 연결이 되었는지 확인합니다.
import redis
# 1. 연결 설정
# Managed Redis는 포트가 10000번이며 ssl=True가 필수입니다.
r = redis.Redis(
host='siren-signup-redis.koreacentral.redis.azure.net',
port=10000,
password='복사한_Primary_Key_값',
ssl=True,
decode_responses=True # 데이터를 문자열로 편하게 읽기 위한 옵션
)
try:
# 2. 연결 확인 (Ping)
if r.ping():
print("✅ Redis 연결 성공! 🚑")
# 3. 테스트 데이터 저장
r.set('connection_test', 'Success')
print(f"테스트 데이터 조회: {r.get('connection_test')}")
except Exception as e:
print(f"❌ 연결 실패: {e}")
회원가입 시 느린 DB 조회를 대신하는 캐싱 코드 작성
그렇다면 이제 Redis를 활용하여 '회원가입 병원 목록'을 캐싱하는 로직을 만들어보겠습니다.
가장 먼저 여러 라우터에서 재사용할 수 있도록 Redis 연결 설정을 별도 파일로 분리합니다.
import redis
import os
from dotenv import load_dotenv
# .env 파일의 내용을 로드합니다.
load_dotenv()
# 환경 변수에서 설정값을 가져옵니다.
REDIS_HOST = os.getenv("REDIS_HOST")
REDIS_PORT = int(os.getenv("REDIS_PORT", 10000)) # 포트는 숫자형이어야 하므로 int() 변환
REDIS_KEY = os.getenv("REDIS_KEY")
# Redis 연결 설정
redis_client = redis.Redis(
host=REDIS_HOST,
port=REDIS_PORT,
password=REDIS_KEY,
ssl=True,
decode_responses=True # 데이터를 문자열로 자동 변환 (bytes -> str)
)
def get_redis():
"""
다른 모듈에서 Redis 클라이언트를 가져올 때 사용하는 함수
"""
return redis_client
위의 redis_config.py를 잘 만들었다면 다름으로 Cache Aside 패턴이 적용된 회원가입 로직 코드를 작성해봅시다.
import bcrypt
import json
import time
from fastapi import APIRouter, HTTPException
from database import get_db_connection
from utils.redis_config import get_redis
router = APIRouter(
prefix="/api/v1/signup",
tags=["Signup"]
)
# 1. 프론트엔드 선택창을 위한 병원 목록 조회
@router.get("/list")
async def get_hospitals():
start_time = time.time()
rd = get_redis()
cache_key = "signup_hospital_list"
# [로그] 함수 호출 확인
print(f"\n📢 [API] 병원 목록 조회 요청 수신")
# 1. Redis에서 데이터 확인
cached_data = rd.get(cache_key)
if cached_data:
duration = time.time() - start_time
# ✅ 캐시 히트 시 로그
print(f"==================================================")
print(f"🚀 [REDIS] Cache Hit! (Redis에서 가져옴)")
print(f"==================================================")
print(f"⏱️ [Redis] 데이터 조회 시간: {duration:.4f}초")
return json.loads(cached_data)
# ❌ 캐시 미스 시 로그 (데이터가 없어서 DB로 가는 상황)
print(f"--------------------------------------------------")
print(f"⚠️ [DATABASE] Cache Miss! (Oracle DB 접근 중...)")
print(f"--------------------------------------------------")
# Redis에 데이터가 없으면 DB에서 조회
connection = get_db_connection()
cursor = connection.cursor()
try:
# 이미 가입된 병원은 제외한 목록
sql = """
SELECT HPID, HOSPITAL_NM
FROM HOSPITAL_MASTER
WHERE HPID NOT IN (SELECT HPID FROM HOSPITAL_LOGIN)
ORDER BY HOSPITAL_NM
"""
cursor.execute(sql)
hospitals = [{"hpid": row[0], "name": row[1]} for row in cursor.fetchall()]
# DB 결과를 Redis에 캐시 (유효기간 1시간)
rd.set(cache_key, json.dumps(hospitals), ex=3600)
duration = time.time() - start_time # 소요 시간 계산
print(f"🏠 [DB -> REDIS 저장완료] 소요 시간: {duration:.4f}초")
return hospitals
finally:
cursor.close()
connection.close()
# 2. 회원가입 처리 (비밀번호 해싱 및 저장)
@router.post("")
async def signup(request: dict):
hpid = request.get("hpid")
password = request.get("password")
if not hpid or not password:
raise HTTPException(status_code=400, detail="HPID와 비밀번호를 모두 입력해주세요.")
# [보안] 비밀번호 해싱 (앞서 사용한 방식과 동일)
salt = bcrypt.gensalt()
hashed_password = bcrypt.hashpw(password.encode('utf-8'), salt).decode('utf-8')
connection = get_db_connection()
cursor = connection.cursor()
try:
# 중복 가입 체크
check_sql = "SELECT COUNT(*) FROM HOSPITAL_LOGIN WHERE HPID = :hpid"
cursor.execute(check_sql, hpid=hpid)
if cursor.fetchone()[0] > 0:
raise HTTPException(status_code=400, detail="이미 가입된 병원입니다.")
# 데이터 삽입
insert_sql = "INSERT INTO HOSPITAL_LOGIN (HPID, PASSWORD) VALUES (:hpid, :password)"
cursor.execute(insert_sql, hpid=hpid, password=hashed_password)
connection.commit() # INSERT 후에는 반드시 commit
# 가입 성공 시 Redis 캐시 삭제
rd = get_redis()
rd.delete("signup_hospital_list")
print("Redis Cache Cleared for signup_hospital_list")
return {"message": "회원가입이 완료되었습니다."}
except Exception as e:
connection.rollback()
print(f"Signup Error: {e}")
raise HTTPException(status_code=500, detail="서버 오류가 발생했습니다.")
finally:
cursor.close()
connection.close()
프론트엔드에서 /signup/list 엔드포인트로 접근해 데이터를 가져갈 수 있도록 라우터를 만듭니다.
Redis의 성능 비교를 위해 시간을 저장합니다.
그리고 앞서 만든 redis_config.py에서 get_redis( ) 함수를 불러와 Redis 서버와의 통신을 연결합니다.
cache_key는 Redis 데이터의 유익한 식별자로 다른 데이터와 섞이지 않게 해줍니다.
@router.get("/list")
async def get_hospitals():
start_time = time.time() # 검색 속도 측정을 위해 시작 시간을 기록합니다.
rd = get_redis() # 메모리 DB인 Redis에 접근하기 위한 클라이언트를 가져옵니다.
cache_key = "signup_hospital_list" # Redis에서 사용할 사물함 번호(키)를 정의합니다.
앞서 발급한 키를 바탕으로 Redis에 데이터가 있는지 get 해봅니다.
만약 데이터가 있다면 해당 데이터를 JSON 객체로 반환합니다.
cached_data = rd.get(cache_key) # 먼저 Redis 사물함에 데이터가 있는지 확인합니다.
if cached_data:
duration = time.time() - start_time # 현재 시간에서 시작 시간을 빼서 걸린 시간을 계산합니다.
print(f"🚀 [REDIS] Cache Hit! (Redis에서 가져옴)")
return json.loads(cached_data) # 문자열로 저장된 데이터를 파이썬 객체로 변환해 즉시 반환합니다.
만약 Redis가 없으면 DB로 접근하여 데이터를 가져옵니다.
# Redis에 데이터가 없으면(Cache Miss) 로그를 찍고 DB로 향합니다.
connection = get_db_connection() # Oracle 또는 메인 DB에 접속을 시도합니다.
cursor = connection.cursor() # SQL 명령어를 전달할 심부름꾼(커서)을 생성합니다.
회원가입 시 보여줄 병원 리스트와 HPID를 DB에서 쿼리해옵니다. 그 내용을 hospitals 변수에 저장합니다.
sql = """
SELECT HPID, HOSPITAL_NM FROM HOSPITAL_MASTER
WHERE HPID NOT IN (SELECT HPID FROM HOSPITAL_LOGIN)
ORDER BY HOSPITAL_NM
"""
cursor.execute(sql) # 작성한 필터링 SQL 쿼리를 실행합니다.
hospitals = [{"hpid": row[0], "name": row[1]} for row in cursor.fetchall()]
이렇게 요청한 DB 결과를 이제는 Redis에 저장하여 다음 번에 접근할 때는 빠르게 보도록 해줍니다.
1시간(3600초) 뒤에 캐시가 자동 삭제되게 하여 최신성을 유지합니다.
# DB 결과를 Redis에 저장하여 다음 사람부터는 빠르게 보게 합니다.
rd.set(cache_key, json.dumps(hospitals), ex=3600)
duration = time.time() - start_time # DB까지 다녀온 총 소요 시간을 계산합니다.
return hospitals # 최종 결과를 반환합니다.
다음의 코드는 앞선 포스팅에서 사용한 회원가입 요청 라우터 기능이고 해당 코드 마지막에 새로운 병원 가입이 완료되었다면 Redis의 정보를 변경해야 함으로 가입된 정보를 delete 합니다.
로그인 시 서버가 발행한 토큰을 관리하는 세션(Session) 관리 코드 작성
다음으로 단순히 토큰을 브라우저에 던져주고 잊어버리는 것이 아니라, Redis라는 실시간 출입 명부에 기록해두는 코드를 작성해봅시다. 이전에 사용한 login.py 코드를 사용합니다.
기존의 login.py 코드에서 로그인을 했을 때 [데이터 조회 - 검증 - 토큰/세션 생성 - 쿠키 발급] 순서로 진행됩니다.
import bcrypt
import os
import json
from datetime import datetime, timedelta, timezone
from jose import jwt
from fastapi import APIRouter, HTTPException, status
from fastapi.responses import JSONResponse
from database import get_db_connection
from utils.redis_config import get_redis
# 로그인 보안설정
SECRET_KEY = os.getenv("SECRET_KEY", "fallback_secret_for-dev")
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60
router = APIRouter(
prefix="/api/v1/login",
tags=["Login"]
)
@router.post("")
async def login(request: dict):
hpid = request.get("hpid")
password = request.get("password")
if not hpid or not password:
raise HTTPException(status_code=400, detail="HPID and Password are required")
connection = get_db_connection()
cursor = connection.cursor()
rd = get_redis()
try:
sql = """
SELECT L.PASSWORD, M.HOSPITAL_NM, L.HPID, M.REGION
FROM HOSPITAL_LOGIN L
JOIN HOSPITAL_MASTER M ON L.HPID = M.HPID
WHERE L.HPID = :hpid
"""
cursor.execute(sql, hpid=hpid)
result = cursor.fetchone()
db_hashed_password = result[0].strip() # 공백 제거 필수!
hospital_name = result[1]
region = result[3]
# bcrypt로 비밀번호 검증 (입력받은 비번과 DB의 해시값을 대조)
if not bcrypt.checkpw(password.encode('utf-8'), db_hashed_password.encode('utf-8')):
return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED,
content={"detail": "Invalid HPID or password"}
)
# 로그인 성공시 JWT 토큰 생성 (토큰에 담을 정보 : HPID, 병원명, 만료시간)
expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
payload = {
"sub": hpid,
"name": hospital_name,
"exp": expire
}
# 페이로드와 비밀키를 조합하여 디지털 서명된 토큰 발행
access_token = jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
# Redis 세션 정보 저장
session_data = {
"hpid": hpid,
"name": hospital_name,
"login_at": datetime.now().isoformat()
}
rd.set(access_token, json.dumps(session_data), ex=ACCESS_TOKEN_EXPIRE_MINUTES * 60)
print(f"[Redis] 세션 저장 완료: {session_data}")
content = {
"message": "Login successful",
"hospital_name": hospital_name,
"region": region,
"hpid": hpid
}
# 쿠리를 수동으로 설정
response = JSONResponse(content=content)
# 쿠키 설정
response.set_cookie(
key="access_token",
value=access_token,
httponly=True, # 브라우저 스크립트 접근 차단 (XSS 해킹 방지)
secure=True, # HTTPS 환경에서만 전송
samesite="lax", # 타 사이트에서 쿠키 전송 제한
max_age=3600 # 쿠키 유효시간
)
return response
except Exception as e:
print(f"Login error: {e}")
raise HTTPException(status_code=500, detail="Internal Server Error")
finally:
cursor.close()
connection.close()
SECRET_KEY는 서버만 알고 있는 '인증 도장'입니다. 이 키가 없으면 누구도 유효한 토큰을 위조할 수 없습니다.
HS256은 가장 대중적이고 안전한 대칭키 암호화 방식입니다.
뒤에 나올 payload 변수와 함께 SECRET_KEY를 담아 토큰을 발행합니다.
# 로그인 보안설정
SECRET_KEY = os.getenv("SECRET_KEY", "fallback_secret_for-dev") # 토큰 변조 방지를 위한 서버 비밀키
ALGORITHM = "HS256" # JWT 암호화 알고리즘 (HMAC + SHA256)
ACCESS_TOKEN_EXPIRE_MINUTES = 60 # 토큰 유효 기간 (1시간)
사용자가 로그인을 할 때 보낸 JSON 본문에서 hpid와 password를 추출하고, 하나라도 없으면 400 에러를 반환하여 불필요한 DB 조회를 방지합니다.
@router.post("")
async def login(request: dict):
hpid = request.get("hpid") # 병원 고유 ID
password = request.get("password") # 입력받은 비밀번호
if not hpid or not password:
raise HTTPException(status_code=400, detail="HPID and Password are required")
DB에 접근해 로그인 테이블에서 비밀번호를 가져오고, 마스터 테이블에서 병원 이름과 지역 정보를 동시에 가져오는 JOIN을 사용했습니다.
connection = get_db_connection()
cursor = connection.cursor()
rd = get_redis() # Redis 클라이언트 가져오기
try:
sql = """
SELECT L.PASSWORD, M.HOSPITAL_NM, L.HPID, M.REGION
FROM HOSPITAL_LOGIN L
JOIN HOSPITAL_MASTER M ON L.HPID = M.HPID
WHERE L.HPID = :hpid
"""
cursor.execute(sql, hpid=hpid)
result = cursor.fetchone()
사용자가 입력한 ID를 가지고 DB에서 해당 사용자의 비밀번호를 가져옵니다.
해시값이 적용된 비밀번호에 bcrypt.checkpw를 사용하여 사용자가 입력한 평문 암호를 해싱하여 대조합니다.
아이디 혹은 비밀번호가 틀리다면 return 값을 반환합니다.
db_hashed_password = result[0].strip() # DB에 저장된 해시값 (공백 제거)
hospital_name = result[1]
# bcrypt로 비밀번호 검증
if not bcrypt.checkpw(password.encode('utf-8'), db_hashed_password.encode('utf-8')):
return JSONResponse(...)
payload 변수는 토큰 안에 담길 데이터입니다. 서버는 나중에 이 토큰만 보고 "아, 이 사람은 어느 병원 사람구나"를 즉시 알 수 있습니다.
또한, 만료 시간을 넣어 토큰 유출 시 피해를 최소화합니다.
# 로그인 성공시 JWT 토큰 생성
expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
payload = {
"sub": hpid, # 토큰 식별자 (Subject)
"name": hospital_name, # 병원 이름 정보 포함
"exp": expire # 만료 시간 설정
}
# 디지털 서명된 토큰 발행
access_token = jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
클라이언트는 JWT를 가지고 있고, 서버(Redis)는 그 토큰이 유효한지 장부에 기록해둡니다. Redis에 기록해두는 과정을 세션이라고 부릅니다.
나중에 특정 사용자를 강제 로그아웃시키고 싶다면, Redis에서 해당 토큰 키만 삭제하면 됩니다. (JWT만 쓰면 불가능한 기능입니다.)
# Redis 세션 정보 저장
session_data = {
"hpid": hpid,
"name": hospital_name,
"login_at": datetime.now().isoformat()
}
# 토큰을 Key로, 정보를 Value로 저장 (만료 시간 설정)
rd.set(access_token, json.dumps(session_data), ex=ACCESS_TOKEN_EXPIRE_MINUTES * 60)
httponly, secure, samesite 설정을 통해 브라우저 환경에서 발생할 수 있는 대부분의 인증 탈취 공격을 방어합니다.
response = JSONResponse(content=content)
# 쿠키 설정
response.set_cookie(
key="access_token",
value=access_token,
httponly=True, # 자바스크립트 탈취(XSS) 방지
secure=True, # HTTPS 통신 강제
samesite="lax",# 가짜 요청(CSRF) 방지
max_age=3600 # 1시간 유지
)'FastAPI' 카테고리의 다른 글
| [FastAPI] JWT를 이용한 회원가입/로그인 서버 구현 (0) | 2025.12.21 |
|---|---|
| [FastAPI] Nginx 사용하기 (0) | 2025.12.18 |
| [FastAPI] Azure Service Bus와 Function App을 활용한 데이터 전송 (2) (0) | 2025.12.18 |
| [FastAPI] Azure Service Bus와 Function App을 활용한 데이터 전송 (1) (0) | 2025.12.18 |
| [FastAPI] Naver API를 사용한 거리 및 최단경로 시간 탐색 (0) | 2025.12.17 |