포트폴리오 최신화하고 코딩테스트 깊게 하기위하여 몇일동안은 기본문제만 진행
포트폴리오
TIL 내용 정리
개선 사항
user
성능 최적화: 이메일 발송 로직의 비동기(Asynchronous) 처리
- 이슈
- password_service.py와 email_service.py의 send_mail(...) 함수가 동기적으로 동작함
- 이메일을 발송하기 위해 구글 SMTP 서버와 통신하는 시간 동안
- 사용자는 회원가입이나 메일 발송 API 응답을 하염없이 기다려야 함 (네트워크 지연 발생)
- 해결 가이드
- Celery + Redis (또는 장고 최신 버전의 async/await 기능)를 도입하여
- 이메일 발송 작업을 백그라운드 워커(Worker)로 넘기기
보안성 개선: 무차별 대입 공격(Brute-Force) 방어 로직 부재
- 이슈
- 현재 코드는 이메일 인증번호 발송 API나 로그인 API에 대해
- 재시도 횟수 제한(Rate Limiting)이 존재하지 않음
- 악의적인 봇이 짧은 시간에 수만 건의 로그인 시도나 이메일 발송을 요청할 경우
- 서버 다운이나 SMTP 쿼터 초과가 발생할 수 있음
- 해결 가이드
- DRF에서 기본 제공하는 Throttling 기능을 도입 추천
가독성 및 예외 처리 구체화
- 이슈
- social_login_service.py에서 requests.post()를 통해
- 외부 서버와 통신할 때 네트워크 오류나 HTTP 상태 코드 에러에 대한 방어가 적음
- 해결 가이드
- requests 라이브러리의 raise_for_status() 메서드나 Timeout 설정을 명시적으로 추가하여
- 외부 API 장애가 우리 서버의 무한 대기(Hang) 현상으로 이어지지 않도록 방어 로직을 리팩토링 하기
무차별 대입 공격(Brute-Force) 방어
- DRF에서 제공하는
ScopedRateThrottle을 사용하면
- 뷰(View) 단위로 아주 쉽게 IP 기반 속도 제한을 걸 수 있음
- 하지만 실무에서 간과하기 쉬운 치명적인 문제가 존재
- 만약 서버가 Nginx, AWS ALB 같은 로드밸런서(리버스 프록시) 뒤에 있다면
- Django가 인식하는 IP가 실제 접속자의 IP가 아니라 로드밸런서의 IP가 될 수 있음
- 이렇게 되면 한 유저가 로그인을 5번 시도했는데 모든 유저가 차단당하는 대참사가 발생 가능
최선의 선택
- DRF 기본 Throttling을 적용하되
- 배포 환경이라면 반드시 settings.py에서
- 클라이언트의 진짜 IP(X-Forwarded-For)를 식별할 수 있도록
- 프록시 관련 설정(NUM_PROXIES 등)을 맞춰주어야 함
settings.py (요금제 설정 추가)
REST_FRAMEWORK = {
'DEFAULT_THROTTLE_CLASSES': [
'rest_framework.throttling.AnonRateThrottle',
'rest_framework.throttling.UserRateThrottle'
],
'DEFAULT_THROTTLE_RATES': {
'email_send': '3/min',
'login_attempt': '5/min',
},
'NUM_PROXIES': 1,
}
apps/user/views/login.py (로그인 뷰 적용)
class LoginAPIView(APIView):
permission_classes = [AllowAny]
serializer_class = LoginSerializer
——————————————————————————————————————[비교]—————————————————————————————————————————
from rest_framework.throttling import ScopedRateThrottle
class LoginAPIView(APIView):
permission_classes = [AllowAny]
serializer_class = LoginSerializer
throttle_classes = [ScopedRateThrottle]
throttle_scope = 'login_attempt'
apps/user/views/email_view.py (이메일 발송 뷰 적용)
class EmailSendView(APIView):
"""이메일 발송 요청을 처리하는 API 뷰"""
permission_classes = [AllowAny]
——————————————————————————————————————[비교]—————————————————————————————————————————
from rest_framework.throttling import ScopedRateThrottle
class EmailSendView(APIView):
"""이메일 발송 요청을 처리하는 API 뷰"""
permission_classes = [AllowAny]
throttle_classes = [ScopedRateThrottle]
throttle_scope = 'email_send'
NUM_PROXIES 값을 정하는 방법
NUM_PROXIES: None (기본값)
- 프록시를 안 쓸 때 (로컬 개발 환경). 쓰로틀링이 접속한 바로 그 IP(공유기 등)를 잡음
NUM_PROXIES: 1
- Django 앞에 Nginx 하나만 있거나, AWS ALB(로드밸런서) 하나만 있을 때. (가장 흔한 배포 환경)
NUM_PROXIES: 2
- 클라이언트 ➡️ AWS CloudFront ➡️ AWS ALB ➡️ Django 처럼 거쳐오는 프록시가 2개일 때
가독성 및 예외 처리 구체화 (외부 API 통신 안정성 확보)
- timeout 설정과 raise_for_status()를 추가
- 서버가 Hang(무한 대기) 상태에 빠지는 것을 막는 핵심
최선의 선택
- 단순히 4xx, 5xx 에러만 잡는 것이 아니라
- 외부 서버 아예 죽어버렸거나(ConnectionError), 5초가 넘어버리는(Timeout) 상황 등
- requests 라이브러리에서 발생할 수 있는 모든 네트워크 관련 예외를 한 번에 묶어서 안전하게 처리
GitHub
class GithubLoginService:
...
token_req = requests.post(token_req_url, data=data, headers=headers)
token_json = token_req.json()
error = token_json.get("error")
if error is not None:
raise ValueError("GitHub 토큰을 받아오는데 실패했습니다.")
access_token = token_json.get("access_token")
user_req_url = "https://api.github.com/user"
user_req = requests.get(
user_req_url, headers={"Authorization": f"Bearer {access_token}"}
)
user_json = user_req.json()
...
——————————————————————————————————————[비교]—————————————————————————————————————————
class GithubLoginService:
...
try:
token_req = requests.post(
token_req_url,
data=data,
headers=headers,
timeout=7
)
token_req.raise_for_status()
token_json = token_req.json()
if "error" in token_json:
raise ValueError("GitHub 토큰을 받아오는데 실패했습니다.")
access_token = token_json.get("access_token")
user_req_url = "https://api.github.com/user"
user_req = requests.get(
user_req_url,
headers={"Authorization": f"Bearer {access_token}"},
timeout=5
)
user_req.raise_for_status()
user_json = user_req.json()
except RequestException as e:
raise ValueError(f"GitHub 서버와 통신 중 지연 혹은 오류가 발생했습니다. (상세: {e})")
class DiscordLoginService:
...
token_req = requests.post(token_req_url, data=data, headers=headers)
token_json = token_req.json()
if "error" in token_json:
raise ValueError("Discord 토큰을 받아오는데 실패했습니다.")
access_token = token_json.get("access_token")
user_req_url = "https://discord.com/api/users/@me"
user_req = requests.get(
user_req_url, headers={"Authorization": f"Bearer {access_token}"}
)
user_json = user_req.json()
——————————————————————————————————————[비교]—————————————————————————————————————————
class DiscordLoginService:
...
try:
token_req = requests.post(
token_req_url,
data=data,
headers=headers,
timeout=7
)
token_req.raise_for_status()
token_json = token_req.json()
if "error" in token_json:
raise ValueError("Discord 토큰을 받아오는데 실패했습니다.")
access_token = token_json.get("access_token")
user_req_url = "https://discord.com/api/users/@me"
user_req = requests.get(
user_req_url,
headers={"Authorization": f"Bearer {access_token}"},
timeout=5
)
user_req.raise_for_status()
user_json = user_req.json()
except RequestException as e:
raise ValueError(f"Discord 서버와 통신 중 지연 혹은 오류가 발생했습니다. (상세: {e})")
AI
보안 및 비용 최적화: 입력값 길이 제한과 Rate Limiting 부재
- 이슈
- 외부 LLM API(Gemini)는 요청한 텍스트의 길이에 비례하여 비용이 발생
- 현재 뷰에서는 사용자가 보낸 text의 길이를 제한하지 않으므로
- 악의적으로 수만 자의 텍스트를 요청하여 API 비용을 폭증시키거나
- 쿼터를 고갈시킬 수 있음
- 또한 무차별 요청(Rate Limiting)에 대한 방어도 없음
- 개선 가이드
- 단순 딕셔너리(request.data.get) 추출 대신 DRF의 Serializer를 도입하여
- serializers.CharField(max_length=2000) 등으로
- 요청 텍스트의 최대 길이를 반드시 제한 필요
- 해당 View에
throttle_classes = [UserRateThrottle, AnonRateThrottle] 를
- 추가하여 하루/분당 API 호출 횟수를 제한 필요
성능 최적화: SDK Client 인스턴스 재사용
- 이슈
- openai_service.py를 보면 convert_text_tone 함수가 호출될 때마다
- 매번 client = genai.Client(...) 객체를 새롭게 생성하고 있음
- 개선 가이드
- 클라이언트 초기화 비용을 줄이기 위해
- 파일 최상단(글로벌 영역)에서 client를 한 번만 초기화해 두고
- 여러 요청이 이를 재사용하도록 수정하는 것이 성능상 유리
- (만약 해당 SDK가 멀티스레드 환경에서 Thread-Safe 하다는 전제 하에 적용)
안정성 개선: 스트리밍 중 예외 처리의 한계
- 이슈
- ToneConverterAPIView의 try-except 블록은
- 초기 연결 실패 시에는 400, 500 에러를 잘 반환함
- 하지만 StreamingHttpResponse가 이미 시작되어
- 클라이언트로 200 OK 상태 코드가 날아간 이후에 텍스트를 생성하다가
- 오류(예: Google 서버 일시 끊김)가 발생하면
- 서버는 상태 코드를 500으로 바꿀 수 없고 그냥 연결을 뚝 끊어버리게 됨
- 개선 가이드
- 프론트엔드가 중간 에러를 명확히 인지할 수 있도록
- 순수 텍스트(문자열 바이트)만 yield 하는 대신 구조화된 SSE(Server-Sent Events) 포맷을 사용
보안 및 비용 최적화
from rest_framework import serializers
class ToneConvertSerializer(serializers.Serializer):
"""
텍스트 변환 API의 입력값을 검증하는 시리얼라이저입니다.
"""
text = serializers.CharField(
max_length=2000,
required=True,
error_messages={
"required": "변환할 텍스트를 입력해주세요.",
"max_length": "텍스트는 최대 2000자까지 변환 가능합니다."
}
)
tone = serializers.CharField(
required=True,
error_messages={
"required": "원하시는 문체를 선택해주세요."
}
)
class ToneConverterAPIView(APIView):
def post(self, request, *args, **kwargs):
text = request.data.get("text")
tone = request.data.get("tone")
if not text or not tone:
raise BaseCustomException(ErrorMessage.INVALID_INPUT)
...
——————————————————————————————————————[비교]—————————————————————————————————————————
from rest_framework.throttling import UserRateThrottle, AnonRateThrottle
class ToneConverterAPIView(APIView):
throttle_classes = [UserRateThrottle, AnonRateThrottle]
def post(self, request, *args, **kwargs):
serializer = ToneConvertSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
text = serializer.validated_data.get("text")
tone = serializer.validated_data.get("tone")
REST_FRAMEWORK = {
...
'DEFAULT_THROTTLE_RATES': {
'anon': '100/day',
'user': '1000/day',
'email_send': '3/min',
'login_attempt': '5/min',
},
성능 최적화: SDK Client 인스턴스 재사용
- Client 글로벌화
- genai.Client는 HTTP 세션을 내부적으로 관리하므로
- 매번 생성하면 불필요한 네트워크 핸드셰이크 시간이 낭비됨
- 전역(Global) 영역에서 한 번만 생성하고 스레드끼리 공유하도록 빼내기
- SSE 에러 처리
- 스트리밍 중에 구글 서버가 죽어버리면 Django는 이미 200 OK를 내보냈기 때문에 500 에러로 바꿀 수 없음
- 따라서 텍스트가 아닌 "구조화된 SSE 이벤트"(data: ..., event: error) 형태로 내려주어
- 프론트엔드가 이를 파싱(Parsing)해 에러를 인지할 수 있도록 함
def convert_text_tone(text: str, tone: str) -> str:
try:
system_prompt = TONE_MAPPING.get(tone)
if not system_prompt:
raise BaseCustomException(ErrorMessage.UNSUPPORTED_TONE)
client = genai.Client(api_key=settings.GEMINI_API_KEY)
response = client.models.generate_content_stream(
model="gemini-2.5-flash",
contents=text,
config=types.GenerateContentConfig(
system_instruction=system_prompt,
temperature=0.7,
max_output_tokens=2500,
),
)
for chunk in response:
if chunk.text:
yield chunk.text.encode("utf-8")
except ValueError as ve:
logger.error(f"잘못된 문체 요청: {ve}")
raise
except Exception as e:
logger.error(f"Gemini API 호출 중 오류 발생: {e}")
raise BaseCustomException(ErrorMessage.AI_CONVERSION_FAILED)
——————————————————————————————————————[비교]—————————————————————————————————————————
try:
gemini_client = genai.Client(api_key=settings.GEMINI_API_KEY)
except Exception as e:
logger.error(f"Gemini Client 초기화 실패: {e}")
gemini_client = None
def convert_text_tone(text: str, tone: str):
try:
system_prompt = TONE_MAPPING.get(tone)
if not system_prompt:
raise BaseCustomException(ErrorMessage.UNSUPPORTED_TONE)
if not gemini_client:
raise RuntimeError("AI 클라이언트를 초기화할 수 없습니다.")
response = gemini_client.models.generate_content_stream(
model="gemini-2.5-flash",
contents=text,
config=types.GenerateContentConfig(
system_instruction=system_prompt,
temperature=0.7,
max_output_tokens=2500,
),
)
for chunk in response:
if chunk.text:
sse_payload = json.dumps({"text": chunk.text}, ensure_ascii=False)
yield f"data: {sse_payload}\n\n".encode("utf-8")
except Exception as e:
logger.error(f"Gemini API 스트리밍 중 오류 발생: {e}")
error_payload = json.dumps({"error": "AI 텍스트 생성 중 일시적인 서버 오류가 발생했습니다."}, ensure_ascii=False)
yield f"event: error\ndata: {error_payload}\n\n".encode("utf-8")
exception 수정
import logging
from rest_framework.views import exception_handler
from rest_framework.response import Response
from apps.core.exceptions.messages import ErrorMessage
logger = logging.getLogger("django")
def custom_exception_handler(exc, context):
response = exception_handler(exc, context)
if response is None:
logger.error(f"[System Error] {exc}", exc_info=True)
return Response(
{
"error_detail": ErrorMessage.SYSTEM_ERROR.message,
"code": ErrorMessage.SYSTEM_ERROR.code,
},
status=ErrorMessage.SYSTEM_ERROR.status_code,
)
custom_data = {
"error_detail": response.data.get("detail", "유효하지 않은 요청입니다."),
"code": getattr(exc, "default_code", "error"),
}
if response.status_code == 400 and "detail" not in response.data:
custom_data["errors"] = response.data
response.data = custom_data
return response
——————————————————————————————————————[비교]—————————————————————————————————————————
import logging
from rest_framework.views import exception_handler
from rest_framework.response import Response
from apps.core.exceptions.messages import ErrorMessage
from apps.core.exceptions.base import BaseCustomException
logger = logging.getLogger("django")
def custom_exception_handler(exc, context):
response = exception_handler(exc, context)
if response is None:
logger.error(f"[System Error] {exc}", exc_info=True)
return Response(
{
"code": ErrorMessage.SYSTEM_ERROR.code,
"message": ErrorMessage.SYSTEM_ERROR.message,
},
status=ErrorMessage.SYSTEM_ERROR.status_code,
)
custom_data = {}
if isinstance(exc, BaseCustomException):
custom_data["code"] = exc.default_code
custom_data["message"] = str(exc.detail)
else:
custom_data["code"] = getattr(exc, "default_code", "error")
if isinstance(response.data, dict) and "detail" in response.data:
custom_data["message"] = str(response.data["detail"])
else:
custom_data["message"] = "유효하지 않은 요청입니다."
if response.status_code == 400 and isinstance(response.data, dict) and "detail" not in response.data:
custom_data["code"] = ErrorMessage.INVALID_INPUT.code
custom_data["message"] = ErrorMessage.INVALID_INPUT.message
custom_data["details"] = response.data
response.data = custom_data
return response
결과
- 커스텀 에러 발생 시 (raise BaseCustomException(ErrorMessage.USER_NOT_FOUND))
{
"code": "user_not_found",
"message": "존재하지 않는 유저입니다."
}
- 뷰에서 serializer.is_valid(raise_exception=True)로 검증 실패 시 (400 에러)
{
"code": "invalid_input",
"message": "유효하지 않은 입력값입니다.",
"details": {
"email": ["유효한 이메일 주소를 입력하십시오."],
"password": ["비밀번호는 8자 이상이어야 합니다."]
}
}
- 코드 버그로 인한 서버 터짐 방어 시 (500 에러)
{
"code": "server_error",
"message": "서버 내부 오류가 발생했습니다."
}