[Redis] Redis 기반 할일 목록 과 채팅 기능 만들어보기!

우선 내가 해야할것!

• 일단 Redis 기반 으로 할일 목록과 채팅기능 제공 풀스택 어플을 만들어보자
• 실시간 기능과 효율적인 데이터 관리를 가능할수 있게 만들어보자

System architecture

• Frontend : React
• Backend: FastAPI
• Database: Redis
• Real-time communication: WebSocket

첫번째로 한것!

• 일단 우선 FAST API 구축해보기
  • 왜 FAST API 인가?
• 우선 python 을 활용해서 API를 구축하고 싶어 python이라는 언어를 사용시 고려해야할 방식은 FAST API, FLASK, Django 정도 가 있는데, Fast API 인 경우 빠르고 간단하게 개발이 가능하고 OpenAPI 와 JSON Schema 을 자동으로 생성해서 API 문서화가 간단하고 직관적이고, 아무래도 간단한 구현이다보니 코드를 작성하는데 복잡하게 잡지 않아도 되며, 코드의 가독성 그리고 타입검사 등 안정적인 코드를 작성할수 있다는 장점이 있다. 무엇보다도 문서의 자동화를 통해 API구성시 자동으로 Swagger UI등 별도의 설정없이 API 문서를 확인하고 테스트가 가능하다는 점에서 좋다고 판단했다. 또한 홋시 몰라 미들웨어도 고려했을때 확장성이 뛰어난 Fast API 가 좋다고 판단하여 FastAPI 를 구현하려고 시작했다.
• 그럼 Fast API 는 어떻게 사용하나?

from fastapi import FastAPI
app = FastAPI()

@app.get("/")
async def root():
return {"message": "Hello World"}
### 또는 다른 get, post, delete 등등으로 구현 가능

• 그럼 FastAPI 어떻게 잘 구동되는지 확인하는가?

uvicorn main:app --reload
INFO: Uvicorn running on [http://127.0.0.1:8000](http://127.0.0.1:8000) (Press CTRL+C to quit)  
INFO: Started reloader process \[28720\]  
INFO: Started server process \[28722\]  
INFO: Waiting for application startup.  
INFO: Application startup complete.

보통 uvicorn main:app --reload 을 api 디랙토리에서 사용하게 된다면 서버가 실행되면서 문제가 있는지 없는지 알려준다.
그리고 CTRL + C 를 이용하여 서버를 종료할수있다.
우선 FAST API 를 구축할때 TODO list 를 구현하기위해 이런식으로 코드를 작성했는데..

from fastapi import APIRouter, HTTPException
from app.redis import get_redis_client
from urllib.parse import unquote
import json

router = APIRouter()

# Redis 클라이언트 초기화
redis_client = get_redis_client()

@router.get("/todos")
def list_todos():
    """
    Redis에서 "todo:*" 패턴에 해당하는 모든 작업을 반환합니다.
    """
    try:
        keys = redis_client.keys("todo:*")
        todos = {key: json.loads(redis_client.get(key)) for key in keys}  # JSON 문자열을 파싱
        return {"todos": todos}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error fetching todos: {str(e)}")


@router.post("/todos/{todo_id}")
def create_todo(todo_id: str, task: str, date: str, goal: str):
    """
    새로운 작업을 Redis에 추가합니다.
    """
    redis_key = f"todo:{todo_id}"
    try:
        if redis_client.exists(redis_key):
            raise HTTPException(status_code=400, detail="Todo already exists")

        todo_data = {
            "task": task,
            "date": date,
            "goal": goal
        }
        redis_client.set(redis_key, json.dumps(todo_data))
        return {"message": "Todo created successfully", "todo_id": todo_id, "data": todo_data}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error creating todo: {str(e)}")

@router.put("/todos/{todo_id}")
async def update_todo(todo_id: str, task: str, date: str, goal: str):
    """
    특정 작업을 업데이트합니다.
    """
    redis_key = todo_id if todo_id.startswith("todo:") else f"todo:{todo_id}"
    try:
        if not redis_client.exists(redis_key):
            raise HTTPException(status_code=404, detail="Todo not found")

        todo_data = {
            "task": task,
            "date": date,
            "goal": goal
        }
        redis_client.set(redis_key, json.dumps(todo_data))
        return {"message": "Todo updated successfully", "todo_id": todo_id, "data": todo_data}
    except Exception as e:
        print(f"Error updating todo: {e}")
        raise HTTPException(status_code=500, detail=f"Error updating todo: {str(e)}")


@router.delete("/todos/{key}")
def delete_todo(key: str):
    """
    Redis에서 특정 작업을 삭제합니다.
    """
    decoded_key = unquote(key)  # URL 디코딩
    try:
        if redis_client.exists(decoded_key):
            redis_client.delete(decoded_key)
            return {"message": f"Todo '{decoded_key}' deleted successfully."}
        raise HTTPException(status_code=404, detail="Todo not found")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error deleting todo: {str(e)}")


@router.get("/debug/redis")
async def debug_redis():
    """
    Redis의 모든 작업 디버깅 데이터를 반환합니다.
    """
    try:
        keys = redis_client.keys("todo:*")
        data = {key: redis_client.get(key) for key in keys}
        return {"debug_data": data}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Error fetching debug data: {str(e)}")

이런런식으로 CRUD(CREATE, READ, UPDATE, DELETE) 작업과 디버깅 정보를 위해 FAST API 엔드 포인트를 구현해보았다.
이때 Redis를 통해 DB (Database)를 사용하여 실시간 동기화 기능과 함께 할일 항복을 저장 관리할수 있도로 만들어 놓았다.

일단 Todo list 를 만들고 싶어 todo.py로 파일을 생성하여, FastAPI 라우팅 시스템, Redis client (database 연결), 그리고 URL decoding(특수 문자 처리) 를 처리 할수 있도록 생각 하여, 할일 목록 조회시 필요한 GET, POST, PUT, DELETE를 사용해 할일을 보고, 할일을 생성하고, 할일을 추가하고 할일을 업데이트 는 함수를 생각했고 Redis의 모든할일 데이터를 조회하기 위해 따로 debug_redis 함수를 만들어 생성했다.

확실히 구현하기전에 어떤 기능이 필요한지 생각을 하고 구현을 시작해서 상당히 쉽게 코드를 작성할수 있게 되었다.
다만 하는도중에 특수문자처리를 고려를 못했어서, 나중에 URL 디코딩을 하기까지 조금 시간이 리서치가 필요했다.
다음은 main.py 나 redis.py 그리고 미들웨어와 다른 구성을 정리해야겠다.