RAG(Retrieval-Augmented Generation)는 LLM의 단점 중 ‘사실 관계 오류 가능성’과 ‘맥락 이해의 한계’ 를 개선하는데 초점을 맞춘 방법

RAG는 LLM에 외부 지식 베이시를 연결하여 모델의 생성 능력과 사실 관계 파악 능력을 향상시키는 기술

사용 방식

1. 외부 지식 활용
   1. 대규모의 구조화된 지식 베이스(예: DB, 위키피디아)를 모델에 연결
   2. 주어진 질의에 대한 관련 정보를 지식 베이스에 검색 및 추출
2. 증거 기반 생성
   1. 검색된 지식 정보를 증거로 활용하여 보다 사실에 기반한 답변 생성
   2. 생성된 답변의 출처를 명시함으로써 신뢰성 향상
3. 맥락 이해력 향상
   1. 외부 지식을 통해 질의에 대한 배경 지식과 맥락 정보를 파악
   2. 단순한 패턴 매칭이 아닌 추론 능력을 바탕으로 한 답변 생성

즉, RAG는 기존 LLM의 생성 능력과 외부 지식 베이스의 정보를 결합함으로써, 보다 정확하고 사실에 기반한 답변을 제공할 수 있음. 또한 모델의 출력결과에 대한 증거를 제시할 수 있어 설명 가능성과 신뢰성을 높일 수 있음

구성 요소

1. 질의 인코더(Query Encoder): 사용자가 질문을 이해가 위한 언어 모델. 주어진 질문을 벡터 형태로 인코더
2. 지식 검색기(Knowledge Retriever): 인코딩된 질문을 바탕으로 외부 지식 베이스에서 관련 정보를 검색
3. 지식 증강 생성기 (Knowledge-Augmented Generator): 검색된 지식을 활용하여 질문에 대한 답변을 생성하는 언어 모델. 기존의 LLM과 유사하지만, 검색된 지식을 추가 입력으로 받아 보다 정확하고 풍부한 답변 생성

RAG의 동작 과정 요약

1. 사용자의 질문이 주어지면 질의 인코더가 이를 이해하기 쉬운 형태로 변환
2. 지식 검색기가 인코딩된 질문을 바탕으로 외부 지식 베이스에서 정보를 검색
3. 검색된 지식은 지식 증강 생성기의 입력으로 전달
4. 지식 증강 생성기는 지식을 활용하여 사용자 질문에 대한 답변 생성

RAG의 등장 배경과 필요성

RAG는 자연어 처리와 인공지능 기술의 발전, 그리고 증가하는 사용자의 요구에 따라 등장하게 되었습니다. RAG의 등장 배경과 필요성을 다음과 같이 정리할 수 있습니다.

1. 지식 기반 질의응답 시스템의 한계
   • 초기의 질의응답 시스템은 주로 제한된 도메인의 구조화된 데이터를 기반으로 동작했습니다. 이는 시스템이 다룰 수 있는 주제와 질문의 유형이 한정적이라는 문제가 있었습니다.
   • 사용자의 다양한 정보 요구를 충족시키기 위해서는 보다 광범위한 지식을 활용할 수 있는 시스템이 필요하게 되었습니다.
2. 비정형 텍스트 데이터의 폭발적 증가
   • 인터넷의 발달과 디지털 기기의 보급으로 웹페이지, 뉴스 기사, 소셜 미디어 게시물 등 비정형 텍스트 데이터가 기하급수적으로 증가하고 있습니다.
   • 이러한 대규모 텍스트 데이터는 방대한 지식을 포함하고 있어, 질의응답 시스템의 지식 베이스로 활용할 수 있는 잠재력이 높습니다.
   • 그러나 비정형 데이터를 효과적으로 처리하고 활용하기 위해서는 기존과는 다른 접근 방식이 필요했습니다.
3. 사전 학습된 언어 모델의 발전
   • BERT, GPT 등 사전 학습된 대규모 언어 모델의 등장은 자연어 처리 분야에 큰 변화를 가져왔습니다.
   • 이러한 언어 모델은 방대한 텍스트 데이터로부터 언어의 구조와 의미를 학습하여, 다양한 언어 이해 및 생성 태스크에서 뛰어난 성능을 보여주었습니다.
   • 사전 학습된 언어 모델을 질의응답 시스템에 활용함으로써, 보다 자연스럽고 문맥을 고려한 답변 생성이 가능해졌습니다.
4. 실시간 정보 제공에 대한 사용자의 요구 증대
   • 인터넷과 모바일 기기의 발달로 사용자들은 언제 어디서나 필요한 정보를 즉시 얻고자 하는 요구가 커지고 있습니다.
   • 단순히 정보를 검색하는 것을 넘어, 대화형 인터페이스를 통해 원하는 정보를 직관적으로 얻고자 하는 사용자가 늘어났습니다.
   • 이에 따라 사용자의 질문을 이해하고 적절한 답변을 실시간으로 제공할 수 있는 지능형 질의응답 시스템의 필요성이 대두되었습니다.
5. 지식 검색과 답변 생성의 통합 필요성
   • 기존의 질의응답 시스템은 지식 검색과 답변 생성을 별도의 단계로 처리하는 경우가 많았습니다. 이로 인해 검색된 정보와 생성된 답변 사이의 정합성이 떨어지는 문제가 발생했습니다.
   • 지식 검색과 답변 생성을 통합적으로 수행할 수 있는 프레임워크의 필요성이 제기되었고, 이는 RAG 아키텍처의 등장으로 이어졌습니다.

RAG 기술 접목 사례

1. Anthropic's Constitutional AI (CAI)
   • Anthropic사는 RAG 기술을 활용한 대화형 AI 모델인 CAI를 개발했습니다.
   • CAI는 대화 과정에서 외부 지식을 활용하여 사용자의 질문에 답변을 생성합니다.
   • 생성된 응답의 근거가 되는 출처를 명시하여 신뢰성을 높였습니다.
2. Perplexity AI
   • Perplexity AI는 RAG 기반의 질의응답 서비스를 제공하는 스타트업입니다.
   • 사용자의 질문에 대해 웹 검색을 통해 관련 정보를 수집하고, 이를 바탕으로 응답을 생성합니다.
   • 제공된 응답의 출처와 검색 과정을 사용자에게 투명하게 공개합니다.

RAG 적용 과정

1. 데이터 준비 및 Chunking

• EDM 예시 보고서, 증빙 문서 수집: 내부 보고서, 정책, 실적 자료 등.
• Chunking(쪼개기):
  • 각 문서를 “논리적 단위(섹션, 헤딩, 표 항목)”로 나누고, 메타데이터(카테고리, 연도,보고서 종류 등)를 붙임

2. 임베딩 및 벡터화

• 임베딩 모델 선정:
  • OpenAI, Cohere, HuggingFace(국문이면 KoSimCSE/BGE-me)
• 벡터 추출:
  • 각 chunk(문단/항목)를 임베딩 모델로 벡터화

3. 벡터 DB 저장

• ChromaDB, Pinecone, mindsDB, Weaviate 등 오픈소스 또는 클라우드 벡터 DB 사용
• 저장 정보:
  • 벡터, chunk의 원문, 메타데이터(특히 heading, 구분, 문서ID)

4. RAG 유형 파이프라인 연결

• 질의(Report 생성 요청) → 벡터화
• 벡터DB에서 의미적으로 가장 유사한 chunk Top-N Retrieval
• Retrieval chunk를 LLM 프롬프트로 증강(augementation)
• LLM(예: GPT-4, Gemini 등)에 전달하여 실제 리포트 생성

샘플 코드 (LlamaIndex + ChromaDB)

아래 코드는 “EDM 예시 보고서”를 RAG 시스템에 저장하고, AI 기반 자동 보고서 생성을 수행하는 과정의 예입니다.

from llama_index.core import SimpleDirectoryReader, VectorStoreIndex, StorageContext, ServiceContext
from llama_index.vector_stores.chroma import ChromaVectorStore
from chromadb.config import Settings
from llama_index.llms.openai import OpenAI

# 1. 문서 로딩 및 파싱
docs = SimpleDirectoryReader("edm_sample_docs/").load_data()

# 2. 벡터DB 연동 및 인덱스 생성
db = chromadb.PersistentClient(path="edm_vectordb")
chroma_store = ChromaVectorStore(chroma_client=db, collection_name="edm_reports")
storage_context = StorageContext.from_defaults(vector_store=chroma_store)

# 3. LlamaIndex로 임베딩 및 저장
index = VectorStoreIndex.from_documents(docs, storage_context=storage_context, service_context=ServiceContext.from_defaults(llm=OpenAI("gpt-3.5-turbo", temperature=0)))

# 4. RAG 질의/리포트 생성
query_engine = index.as_query_engine()
answer = query_engine.query("2024년 우리 회사 환경 실적 보고서를 써줘.")
print(answer)

파이프라인 요약

1. EDM 문서 수집 → 논리적 chunk로 분해
2. 각 chunk를 임베딩(벡터화) 및 벡터DB 저장
3. 리포트 요청(질의) → 벡터화 → 유사 chunk top-N 검색
4. 검색 chunk + 질문 조합하여 LLM 프롬프트 구성
5. LLM에게 결과 생성 요청 → 리포트 생성 및 사용자 제공

RAG 프레임워크

LlamaIndex

특징

• RAG(검색증강생성) 특화: LLM과 외부 데이터를 효율적으로 연결해, 인덱싱·검색·프롬프트 증강을 자동화합니다.
• 여러 데이터 소스 통합: API, DB, 파일(PDF, 워드 등), S3, 구글드라이브, SharePoint 등 다양한 커넥터를 지원.
• 고급 인덱스 설계: 벡터·트리·리스트·키워드 인덱스 등 데이터 특성에 맞게 다양한 구조 설계.
• 빠르고 직관적: 몇 줄의 파이썬 코드만으로 PoC~프로덕션까지 구현 가능, ramp-up(도입 장벽)이 낮음.
• 풍부한 확장성: Prompt, LLM, Embedding 모델을 쉽게 교체·조합, LangChain 등 다른 프레임워크와 연동 가능.

주요 기능 & 사용 흐름

1. 데이터 수집/파싱(Ingestion)
   • 내부 문서, DB, 웹 등에서 데이터 수집 및 자동 파싱.
2. 청킹(Chunking) & 인덱싱
   • 문서를 논리 단위 단락(chunk)별로 분할, 메타데이터와 함께 인덱스 형성.
3. 임베딩(Embedding)
   • 각 chunk를 임베딩 모델로 벡터화 후 벡터DB에 저장.
4. 검색 & 프롬프트 증강
   • 사용자 질의를 임베딩해 유사 chunk를 검색 후, 프롬프트에 증거(문맥)로 자동 조합.
5. LLM 응답
   • 프롬프트를 LLM에 전달, 증빙 기반의 정확한 생성 결과 제공.

비용 구조

• LlamaIndex 자체: 오픈소스 버전은 무료, 관리형 SaaS 플랫폼은 크레딧 기반(Free~Enterprise, 2025년 기준).
  • Free(10,000크레딧/월), Starter, Pro, Enterprise 등 플랜 선택 가능.
  • 추가 사용분은 $1/1,000 크레딧 기준 별도 과금.
• LLM/Embedding 호출 비용: 실제 비용은 사용 LLM(OpenAI, Gemini 등)·Embedding API 비용이 별도 발생.
  • 예) GPT-3.5-turbo는 1,000토큰 당 $0.002
  • 인덱스 구축 시에도 일부 인덱스(TreeIndex 등)에 LLM 호출 필요.
• 비용 예측: 개발 단계에서 토큰 예측기/MockLLM으로 예상 비용 시뮬레이션 가능

난이도

• 도입/학습 곡선:
  • RAG 관점에서는 매우 쉽고, 튜토리얼·샘플코드가 많음.
  • 커스텀 인덱스 설계, 대규모 데이터/멀티소스 통합, 성능 튜닝 등은 중급 이상의 숙련 필요.
• 운영 난이도:
  • 수백만 단위 대규모 데이터에도 안정적으로 확장(스케일), 파이프라인 관리 용이.

강점

• 특화된 RAG 성능: 대규모 데이터(수천~수억 문서)에서도 빠르고 정확한 검색·응답.
• 높은 확장성/유연성: 거의 모든 LLM, 벡터DB, 프레임워크와 연동(Plug&Play).
• 실패/재현 방지: 프롬프트 증강·메타데이터 활용 등 고급기능 내장.
• 커뮤니티 성장: 빠른 기능 개선, 다양한 실전 예제와 문서 지원.

단점 및 한계

• 범용 에이전트 프레임워크(X): LangChain만큼의 '체인/에이전트/워크플로우' 복잡 응용은 다소 제한적.
• 복잡한 커스텀화 시 곡선↑: 고급 인덱스 튜닝, 멀티쿼리, 구조화 데이터 지원 등은 내부 구조 숙지가 필요할 수 있음.
• 소규모 프로젝트에 과할 수 있음: 소규모(몇 페이지)라면 인덱스 구축이 오버헤드.
• 에코시스템 상대 열세: 최대 커뮤니티 규모는 LangChain/LangGraph 등이 더 큼.
• 엔터프라이즈 운영: 대용량+보안+고가용성 요구 시 지속적인 관리·업데이트 필요.

실제 적용 시 TIP

• ESG/EDM에 최적: PDF, 정책문서, 실적보고 등 구조화+비정형 혼합데이터의 증거기반 보고서 자동화에 강력.
• PoC~프로덕션 단계별 사용: 무료 플랜→유료 크레딧 방식으로 확장 무난.
• 데이터 커넥터(LlamaHub) 적극 활용: 내부 ERP/DB~클라우드 스토리지와 손쉽게 연동 가능.

결론:

LlamaIndex는 대용량, 복잡 문서 기반의 AI 검색/생성/RAG 솔루션에 크게 특화된 데이터-LLM 연동 프레임워크입니다.

학습 난이도 대비 강력한 기능, 저렴하거나 예측 가능한 비용, 뛰어난 유연성을 바탕으로 ESG 데이터관리·보고서 자동화 등 EDM 업무에 매우 적합합니다.

PoC~대규모 서비스까지, 실제 데이터를 중심으로 한 실전 최적화가 가능합니다.

LangChain

특징

• 목적:

  ChatGPT, GPT-4 등 LLM을 바탕으로 복잡한 프로세스(데이터 수집→임베딩→검색→생성...)를 쉽고 유연하게 연결하는 프레임워크

• 체인(Chain) 구조:

  “단계별 작업(문서 쪼개기→임베딩→DB검색→생성)” 등을 한데 묶어 순차적으로 실행.

  단계 간 데이터 전달 및 후처리 정의가 명확.

• 에이전트(Agent):

  LLM이 상황에 맞춰 다양한 도구(검색API, 계산기, DB, 코드실행…)를 동적으로 호출

  예: “웹에서 자료 찾고, 요약한 뒤, 엑셀 파일로 출력”

• 방대한 연결성:

  • 30여개 이상 LLM(OpenAI, Anthropic, Google 등) 지원
  • 각종 벡터DB, 일반DB, 클라우드, PDF/웹/Slack/GoogleSheet 등 플러그인 형태로 손쉬운 연동
  • 데이터 파이프라인, 챗봇, RAG 시스템, 자동화 워크플로우 등 테스트→서비스 스케일 확장

• 커뮤니티·예제·확장성:

  가장 많은 사용자와 빠른 업데이트, 사용사례 및 튜토리얼 풍부

  오픈소스+커머셜(사업자용) 모두 지원

비용 구조

• 자체 사용비:

  오픈소스 프레임워크 자체는 무료

  단, 실제 LLM/임베딩 모델/서드파티 툴(벡터DB 등)의 API 사용에 따라 별도 비용 발생

• 운영비용:

  코드 구조에 따라 API 호출 반복, 외부 시스템(검색/DB 트래픽 등)에 따른 직접비가 늘 수 있음

  대량 서비스/멀티에이전트 설계 시 토큰 소모 예측 필요

난이도 및 적용성

• 학습 난이도:

  구조 이해(체인/에이전트/메모리)를 한 번 익히면 매우 유연

  복잡한 업무(질문→검색→로그 저장→후처리 등)를 한 곳에서 통합 관리 가능

  입문 튜토리얼/실무 예제가 매우 많음

• 확장성:

  신규 API, DB, LLM 등 추가 연결/교체가 쉬움

  수십 개의 체인, 자동화 플로우도 일관되게 설계 가능

강점

• 모듈/워크플로우 설계 유연성:

  다양한 데이터/툴/LLM/후처리를 단일 레시피로 손쉽게 조합

• Agent(에이전트) 방식을 통한 동적 작업:

  사용자의 질문과 상황에 따라 자동으로 “웹검색→QR코드 생성→메일 발송”도 현업에 적용 가능

• 방대한 벤더·도구·API 지원:

  OpenAI, Google, Slack, Pinecone, Chroma, Salesforce 등 수십종 쉽게 통합

• 대형 프로젝트, 복합 서비스에 적합:

  (QA+챗봇+자동 문서화+다국어 번역+코드실행 등 동시에 요구시)

단점 및 한계

• 복잡한 구조의 오버헤드:

  단일 기능만 쓴다면 구조가 과할 수 있음(학습/코드 길이↑)

• 디버깅/모니터링:

  체인이 길어지거나 Agent가 동적으로 움직일 때, 중간과정 추적이 불편한 편

• 실행 환경/의존성:

  꼭 필요한 최소 모듈만 불러오는 것이 중요.

  (기본 설치 시 의존성 패키지가 많아질 수 있음)

• 성능:

  워크플로우가 복잡할수록 실제 응답 속도는 간단한 파이프라인 대비 느려질 수 있음

실제 적용 팁

• 단순 RAG만 할 경우 LlamaIndex가 더 쉽고 빠르지만,

  여러 데이터, LLM, 워크플로우·자동화 까지 필요하다면 LangChain이 필수

• 여러 LLM/벡터DB/Agent를 자유롭게 조합해 데모~서비스까지 일관 설계 가능

• ESG/EDM 분야:

  • 외부 벤치마킹 연동, 업무별 자동화(예: ESG 공시 대응 챗봇, 다중 평가보고서 자동화 등)
  • 보고서 생성→파일화→이메일 발송까지 전 프로세스 일관 설계

비교 요약

정리:

• 정교한 문서 검색·증거 반영에 집중: → LlamaIndex
• 여러 도구·워크플로우 체인, Agent 동작 등 복합 자동화: → LangChain

GPT는 현재 가장 널리 알려진 고성능 범용 모델입니다.

특징:

뛰어난 범용성: 추론, 코딩, 글쓰기, 창작 등 거의 모든 분야에서 매우 높은 성능을 보입니다.

강력한 멀티모달: 텍스트뿐만 아니라 이미지, 음성, 비디오를 실시간으로 이해하고 상호작용하는 능력이 뛰어납니다.

방대한 생태계: 가장 많은 사용자와 API를 보유하고 있어 관련 자료나 서드파티 도구를 찾기 쉽습니다.

컨텍스트 윈도우: 128,000 토큰

추천 용도: 복잡한 문제 해결, 전문적인 글쓰기, 코딩 지원 등 최고 수준의 성능이 필요한 대부분의 작업에 적합합니다.

주요 AI 모델별 특징과 컨텍스트 윈도우 크기를 간략하게 정리해 드릴게요.

🤖 OpenAI - GPT 시리즈 (GPT-4o)

GPT-4o는 현재 가장 널리 알려진 고성능 범용 모델입니다.

특징:

뛰어난 범용성: 추론, 코딩, 글쓰기, 창작 등 거의 모든 분야에서 매우 높은 성능을 보입니다.

강력한 멀티모달: 텍스트뿐만 아니라 이미지, 음성, 비디오를 실시간으로 이해하고 상호작용하는 능력이 뛰어납니다.

방대한 생태계: 가장 많은 사용자와 API를 보유하고 있어 관련 자료나 서드파티 도구를 찾기 쉽습니다.

컨텍스트 윈도우: 128,000 토큰

추천 용도: 복잡한 문제 해결, 전문적인 글쓰기, 코딩 지원 등 최고 수준의 성능이 필요한 대부분의 작업에 적합합니다.

🧠 Google - Gemini 시리즈

구글의 서비스와 강력하게 통합된, 특히 대용량 컨텍스트 처리에 특화된 모델입니다.

특징:

초대형 컨텍스트 윈도우: 한 번에 매우 많은 양의 정보를 처리할 수 있어 긴 문서, 코드베이스, 비디오 전체를 분석하고 질문에 답하는 데 독보적입니다.

네이티브 멀티모달: 설계부터 텍스트, 이미지, 음성, 비디오를 통합적으로 이해하도록 만들어져 여러 형태의 데이터를 넘나드는 작업에 강합니다.

구글 생태계 연동: 구글 검색, 워크스페이스(Docs, Sheets 등)와 긴밀하게 연동되어 최신 정보 검색 및 요약 능력이 우수합니다.

컨텍스트 윈도우: 1,000,000 토큰 (기본)

추천 용도: 수백 페이지 분량의 논문·법률 문서 요약, 장시간 비디오 내용 분석, 방대한 코드베이스 이해 등 대용량 정보 처리 작업에 가장 강력합니다.

📚 Anthropic - Claude시리즈

안정성과 긴 컨텍스트 처리 능력의 균형을 맞춘 모델입니다.

특징:

안정성과 윤리성: 유해하거나 위험한 답변을 생성하지 않도록 설계된 '헌법 AI(Constitutional AI)'를 기반으로 하여 기업 환경에서 사용하기에 안정적입니다.

긴 글 요약 및 작성: 긴 컨텍스트 윈도우를 바탕으로 장문의 문서를 정확하게 요약하거나 창의적인 글을 자연스럽게 작성하는 능력이 뛰어납니다.

빠른 응답 속도: 모델 라인업(Haiku, Sonnet, Opus)이 다양하여, 비용과 속도에 맞춰 최적의 모델을 선택할 수 있습니다.

컨텍스트 윈도우: 200,000 토큰

추천 용도: 긴 보고서나 책을 요약하고 분석하는 작업, 안전성이 중요한 고객 응대 챗봇, 창의적인 글쓰기 등에 적합합니다.

LangSmith는 대규모 언어 모델(LLM) 기반 애플리케이션의 개발 주기를 가속화하고, 특히 성능을 체계적으로 평가하고 개선하기 위해 설계된 강력한 프레임워크입니다.

이 문서는 ESG 설문 검수 AI 시스템에 LangSmith를 도입한 이유, 주요 기능, 실제 활용 방안, 도입 성과를 설명합니다.

⸻

💡 2. LangSmith 시스템 개요

LangSmith는 LLM 기반 애플리케이션의 개발, 디버깅, 테스트, 평가 과정을 지원하는 포괄적인 플랫폼입니다.

2.1. 핵심 기능 • 실시간 추적 (Real-time Tracing): LLM 호출, 체인 실행, 에이전트 동작 등을 추적하여 병목 진단 가능 • LLM-as-Judge: LLM이 AI 응답을 평가하는 자동화된 지능형 평가 시스템 • 데이터 관리 (Data Management): 데이터셋과 평가 결과를 저장, 관리, 버전 관리 • 대시보드 & 시각화: 평가 결과를 실시간 시각화 • 협업 기능: 팀원 간 결과 공유 및 피드백 용이

2.2. 주요 평가 방식 • LLM Judge: GPT-4o 등 강력한 모델을 활용해 다면적 기준으로 평가 • 구조화된 피드백: JSON 기반의 상세한 평가 및 개선 제안 제공 • 다중 평가자: 형식, 내용, 논리성 등 다각도로 평가 • 자동 데이터 추적: 모든 실행/결과 기록 저장

⸻

✨ 3. 주요 장점: LLM-as-Judge를 통한 지능형 평가

3.1. 지능형 평가 시스템 구현 • 의미 기반 평가: 키워드가 아닌 문맥/의미 이해 중심 • 도메인 전문성 반영: ESG 전문가처럼 평가 프롬프트 설계 • 구체적 피드백 제공: JSON 형태의 개선 제안 포함 • 일관성 있는 반복 평가: 동일 프롬프트 기준으로 성능 비교 용이

3.2. 기존 방식과의 비교

✅ LLM-as-Judge 평가 프롬프트 예시

judge_prompt = """
[Role]: 당신은 ESG 보고서 전문가이자 엄격한 평가관입니다.
[Task]: 제공된 ESG 섹션 내용을 평가하고 점수 및 개선안을 JSON으로 작성하세요.
[Evaluation Criteria]:
1. 섹션 구조의 논리성 (40점)
2. 표 형식 완성도 (35점)
3. 가독성과 표현 명확성 (25점)

[Output Format]:
{
  "total_score": float,
  "feedback_summary": "전반 요약",
  "criteria_scores": {
    "section_structure_logic": float,
    "table_format_completeness": float,
    "overall_readability": float
  },
  "improvement_suggestions": {
    "section_structure": [...],
    "table_format": [...],
    "readability": [...]
  }
}
"""

🗃️ 4. 데이터 관리 기능

4.1. 체계적인 데이터셋 관리의 이점 • 자동 데이터셋 생성: PoC 로그에서 입력/출력 추출 • 버전 관리: 데이터셋 이력 추적 • 결과 연결: 입력, 출력, 평가 결과를 LangSmith에 자동 저장 • 재사용성: 다양한 테스트에 반복 활용 가능

4.2. Python 예시

from langsmith import Client

class LangSmithDatasetManager:
    def __init__(self, api_key: str):
        self.client = Client(api_key=api_key)
        self.poc_logs = [...]  # 실제 로그 로딩 생략

    def create_langsmith_dataset(self, dataset_name: str, description: str = None):
        examples = []
        for log in self.poc_logs:
            example = {
                "inputs": {"question": log["question_number"]},
                "outputs": {"ground_truth": log["raw_model_answer"]}
            }
            examples.append(example)
        return self.client.create_dataset(
            dataset_name=dataset_name,
            description=description,
            examples=examples
        )

📊 5. 대시보드 및 시각화

5.1. 주요 기능 • 표 형태 결과 시각화 • 실시간 업데이트 • 드릴다운 분석 가능 • 비교 분석 기능 • 필터링 및 검색 기능

5.2. 예시 (가상)

🛠️ 6. ESG 시스템에서의 LangSmith 활용

6.1. 평가 방법론

a. 형식 평가 (llm_format_judge) • 평가 기준: 섹션 구성, 표 정렬, 구조 일관성 등

b. 내용 평가 (llm_content_judge) • 평가 기준: 정책 정보의 정확성, 검수 논리성 등

c. 데이터 완성도 평가 (llm_completeness_judge) • 평가 기준: 필수 데이터 포함 여부, 표 내용 정확성

6.2. 실제 구현 예시

from langsmith import evaluate

class ESGEvaluationSystem:
    def run_evaluation(self, dataset_name):
        results = evaluate(
            self.run_esg_model,
            data=dataset_name,
            evaluators=[
                self.llm_format_judge.evaluate_fn,
                self.llm_content_judge.evaluate_fn,
                self.llm_completeness_judge.evaluate_fn
            ],
            experiment_prefix="esg_evaluation"
        )
        return results

✅ 7. 구현 결과 및 성과

7.1. 테스트 결과

7.2. 도입 효과 • 지능형 평가: 의미 기반 피드백 제공 • 데이터 재현성 확보: 자동화된 버전 관리 • 피드백 루프 단축: 실시간 결과 확인 • 협업 용이: 대시보드를 통한 결과 공유

⸻

🎯 8. 결론: ESG AI 품질 향상의 핵심 도구

LangSmith는 다음을 가능하게 합니다: • LLM-as-Judge를 통한 의미 기반 평가 자동화 • 데이터셋 및 결과의 체계적 관리 • 직관적이고 빠른 대시보드 피드백 • 개발 주기 단축과 협업 증진

👉 ESG AI 시스템에 있어 LangSmith는 필수적 품질 평가 프레임워크입니다.

소프트웨어 디자인 패턴 중 하나인 싱글톤 패턴(Singleton Pattern)은 특정 클래스의 인스턴스, 즉 객체가 애플리케이션 내에서 단 하나만 생성되는 것을 보장하는 설계 방식입니다. 생성자가 여러 번 호출되더라도 실제로 생성되는 객체는 최초에 생성된 하나이며, 이후의 모든 호출에서는 최초 생성된 객체를 반환합니다.

이는 마치 우리 반에 반장은 단 한 명만 존재하는 것과 같습니다. 누구든 "반장 나와!"라고 외치면 항상 같은 그 한 명의 반장이 나오는 것과 같은 원리입니다.

왜 싱글톤 패턴을 사용할까요? 싱글톤 패턴은 주로 다음과 같은 목적을 위해 사용됩니다.

메모리 낭비 방지: 객체를 생성할 때마다 메모리를 할당받게 됩니다. 만약 여러 곳에서 동일한 역할을 하는 객체가 필요할 때마다 새로운 객체를 생성한다면 불필요한 메모리 소모가 발생합니다. 싱글톤 패턴을 사용하면 최초 한 번만 객체를 생성하고 이후에는 계속해서 재사용하므로 메모리를 절약할 수 있습니다. 데이터 공유 용이: 단 하나뿐인 객체이므로, 애플리케이션의 여러 부분에서 해당 객체의 데이터를 공유하고 상태를 일관성 있게 관리하기 용이합니다. 예를 들어, 애플리케이션 전체의 환경 설정을 관리하는 객체나, 데이터베이스 연결을 관리하는 커넥션 풀(Connection Pool) 같은 경우에 유용하게 사용될 수 있습니다. 전역적인 접근성: 싱글톤 객체는 전역 변수처럼 애플리케이션의 어느 곳에서나 쉽게 접근하여 사용할 수 있습니다. 이를 통해 다른 객체들과의 상호작용이 편리해집니다.

싱글톤 패턴의 핵심 구현 원리

싱글톤 패턴을 구현하기 위해서는 다음의 두 가지 핵심 요소를 지켜야 합니다.

1. private 생성자: 외부에서 new 키워드를 사용하여 무분별하게 새로운 객체를 생성하는 것을 막기 위해 생성자의 접근 제어자를 private으로 선언합니다.
2. static 메서드와 static 변수: 유일한 인스턴스를 저장하기 위한 static 변수를 클래스 내부에 선언하고, 이 인스턴스에 접근할 수 있도록 public static 메서드(일반적으로 getInstance()라는 이름을 사용)를 제공합니다. 이 메서드는 인스턴스가 아직 생성되지 않았다면 새로 생성하여 반환하고, 이미 생성되어 있다면 기존의 인스턴스를 반환합니다.

파이썬 구현 예시

class Singleton:
    _instance = None  # 유일한 인스턴스를 저장할 클래스 변수

    def __new__(cls):
        # 인스턴스가 아직 없으면 새로 생성
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            print("싱글톤 객체를 새로 생성했습니다.")
        else:
            print("기존 싱글톤 객체를 반환합니다.")
        return cls._instance

    def __init__(self):
        # __init__은 __new__ 호출 후 매번 호출되므로, 
        # 실제 초기화는 한 번만 수행되도록 주의해야 합니다.
        if not hasattr(self, '_initialized'): # 초기화 플래그를 사용하여 한 번만 초기화되도록
            self.value = "기본값"
            self._initialized = True
            print("싱글톤 객체를 초기화했습니다.")

    def do_something(self):
        print(f"싱글톤 객체에서 작업 수행 중입니다. 현재 값: {self.value}")

# 사용 예시
print("--- 첫 번째 객체 생성 시도 ---")
s1 = Singleton()
s1.value = "첫 번째 설정값"
s1.do_something()

print("\n--- 두 번째 객체 생성 시도 ---")
s2 = Singleton()
s2.do_something()

print("\n--- 세 번째 객체 생성 시도 ---")
s3 = Singleton()
s3.value = "세 번째 설정값 (s1, s2와 동일한 객체)"
s3.do_something()

print("\n--- 객체 동일성 확인 ---")
print(f"s1과 s2는 같은 객체인가요? {s1 is s2}")
print(f"s2와 s3는 같은 객체인가요? {s2 is s3}")
print(f"s1.value: {s1.value}, s2.value: {s2.value}, s3.value: {s3.value}")

코드 설명:

• _instance = None: 클래스 변수로, 싱글톤 인스턴스를 저장합니다. 초기에는 None으로 설정합니다.
• __new__(cls):
  • if cls._instance is None:: 만약 _instance가 아직 생성되지 않았다면 (즉, 첫 번째 객체 생성 시도라면)
  • cls._instance = super().__new__(cls): 부모 클래스의 __new__ 메서드를 호출하여 실제 객체를 생성하고 _instance에 저장합니다.
  • else: 이미 _instance가 존재하면 기존의 인스턴스를 반환합니다.
• __init__(self): __new__는 객체를 생성하고 반환하지만, __init__은 객체가 반환된 후 매번 호출됩니다. 따라서 _initialized와 같은 플래그를 사용하여 실제 초기화 로직은 한 번만 실행되도록 하는 것이 중요합니다.

싱글톤 패턴의 장단점

장점

• 메모리 효율성: 단 하나의 인스턴스만 사용하므로 메모리 낭비를 줄일 수 있습니다.
• 데이터 일관성: 전역적으로 하나의 상태를 유지하므로 데이터의 일관성을 보장하기 쉽습니다.
• 전역 접근: 어디서든 쉽게 접근하여 사용할 수 있어 편리합니다.

단점

• 의존성 증가: 싱글톤 객체를 사용하는 다른 클래스들과의 결합도(Coupling)가 높아져 유연성이 떨어질 수 있습니다.
• 테스트의 어려움: 전역 상태를 가지므로 단위 테스트(Unit Test)를 수행하기가 까다롭습니다. 테스트마다 독립적인 환경을 구축하기 어렵기 때문입니다.
• 단일 책임 원칙 위배 가능성: 하나의 객체가 너무 많은 역할과 데이터를 가지게 될 수 있습니다.
• 멀티스레드 환경에서의 동시성 문제: 여러 스레드가 동시에 getInstance() 메서드를 호출할 경우, 여러 개의 인스턴스가 생성될 수 있는 문제가 발생할 수 있습니다. 이를 방지하기 위한 동기화 처리가 필요합니다.

patch(target, new_callable = AsyncMock)

• 여러개 사용 가능
• 순서 중요 마지막 patch 가 첫번째 인자임 아래에서 위로

@patch("app.repo.get_user", new_callable=AsyncMock)
@patch("app.repo.get_post", new_callable=AsyncMock)
async def test_service(mock_get_post, mock_get_user):  # 마지막 patch가 첫 번째 인자
    ...

.return_value

• mock 함수가 await 되었을 때 리턴해줄 값
• 즉 대체한 함수가 결과값을 어떤 것을 줄지 내가 결정하는 것이다

@patch("app.repo.get_user", new_callable=AsyncMock)
async def test_service(mock_get_user):
    mock_get_user.return_value = User(id=1, name="테스트")
    result = await get_user()  # 이 get_user는 실제 DB 안 타고 mock됨

.side_effect

• 특정 값을 돌려주고싶다 -> 가능 (리스트로도 가능)
• 호출 시 에러 던지고 싶다 -> 가능 (return_value 는 객체를 주는거지 실제 raise 처럼 동작을 안함)
• 여러번 호출시 다른값 -> 리스트 형태 가능

# 1. 예외 발생 (주로 try/except 처리 테스트 시)
mock_get_user.side_effect = Exception("DB 에러")

# 2. 여러 값 순차 리턴
mock_get_user.side_effect = [User(id=1), User(id=2), User(id=3)]

# 3. 일반 return_value 처럼 사용할 떄
mock_get_user.side_effect = lambda: User(id=1)

• side_effect는 다음 중 하나여야 함:
  1. 예외 객체 → side_effect = ValueError("에러") → raise ValueError(...)처럼 작동
  2. 함수 or lambda → side_effect = lambda: User(id=1) → 직접 실행 후 결과 리턴
  3. iterable (list 등) → side_effect = [User(id=1), User(id=2)] → 순서대로 리턴

assert_called_once()

이 mock 함수가 정확히 한 번 호출됐냐? 검사
로직 안에서 조건 분기에 따라 특정 함수가 안 불리거나, 정확히 한 번만 불리는 걸 보장해야 할 때.

assert_called_with(...)

필요한 경우
로직에서 특정 파라미터로 정확하게 함수 호출했는지 확인하고 싶을 때
특히 “값에 따라 다르게 동작”하는 로직에서는 매우 유용

예시

@pytest.mark.asyncio
@patch("app.domain.free_board.services.create_free_board_by_id", new_callable=AsyncMock)
async def test_create_free_board_by_id_service(mock_create):
    # given
    dummy_user = object()
    dummy_data = AsyncMock()
    dummy_result = FreeBoardResponseDTO(
        id=1,
        user=UserSchema(
            id=1
        ),
        title="test title",
        content="test content",
        image_url=None,
        view_count=1,
        created_at="2025-04-23T12:00:00",
        updated_at="2025-04-23T12:00:00",
    )
    mock_create.return_value = dummy_result # 여기서 create_free_board_by_id 이 함수 호출시 값을 지정해줌

    # when
    result = await create_free_board_by_id_service(dummy_data, dummy_user)

    # then
    mock_create.assert_called_once_with(dummy_data, dummy_user) # 그래서 서비스 로직을 잘 작성 하였으면 return 값이 안에서 호출한 저값에대한 mock 데이터 값이 나옴
    assert result == dummy_result

실제 사용자가 대용량으로 접속했을 때, 서버가 잘 버티는지 테스트하기 위해

현재 내 프로젝트는 EC2 micro 서버에서 작동 중이기에 테스트 하기에 적합하지 않고 로컬에서만 실험해 보려한다.

Locust의 역할 = 부하 테스트(Load Testing)

• 수천 명의 사용자가 동시에 요청을 보낼때
• 서버의 응답속도, 실패율, 에러발생 여부, 병목 지점을 추측할 수 있다.

장고에 적용시키기
pip install locust

• locustfile.py 생성

  # locustfile.py
  from locust import HttpUser, between, task
  
  

class UserBehavior(HttpUser): host = "http://127.0.0.1:8000" wait_time = between(1, 3) # 요청 간 간격 (초)

def on_start(self):
    response = self.client.post(
        "/api/user/login/",
        json={"email": "test@example.com", "password": "!!test1234"},
    )

    data = response.json()
    self.access_token = data.get("access_token")
    self.headers = {"Authorization": f"Bearer {self.access_token}"}

@task
def get_profile(self):
    self.client.get("/api/user/profile/", headers=self.headers)

@task
def update_profile(self):
    self.client.patch(
        "/api/user/profile/",
        headers=self.headers,
        json={"nickname": "updated_nick"},
    )

@task
def logout(self):
    self.client.post("/api/user/logout/", headers=self.headers)

* 실행
`locust -f locustfile.py`

## Locust로 테스트하면 좋은 API 유형

| 유형 | 설명 | 예시 |
|------|------|------|
| **트래픽이 몰리는 대표 API** | 사용자 요청이 집중되는 핵심 기능 | 로그인, 메인 페이지, 피드 조회 등 |
| **자원 소모가 큰 API** | 처리 시간/메모리/CPU 부하가 큰 작업 | AI 응답 생성, 파일 업로드/변환 등 |
| **DB에 쓰기 많은 API** | 동시에 많은 사용자가 데이터를 생성/수정 | 댓글 작성, 게시글 등록 등 |
| **외부 연동이 포함된 API** | 외부 API 호출로 지연이나 실패 가능 | 결제, AI API 호출, 이메일 전송 등 |
| **비동기 처리 확인용 API** | 비동기 작업 완료 시간이나 큐 처리 확인 | Celery 작업, AI 응답 대기 등 |


### Django 테스트 코드 vs Locust 테스트 비교표

| 항목              | Django 테스트 코드 (`TestCase`) | Locust 테스트               |
|-------------------|----------------------------------|-----------------------------|
| **목적**           | 기능 정확성 테스트 (유닛/통합)       | 성능, 부하 테스트              |
| **언제 실행?**      | CI/CD, PR 머지 시 자동 실행         | 서버 성능 검증 시 수동/자동 실행 |
| **중점**           | 기능이 정상적으로 동작하는가?        | 얼마나 많은 요청을 견디는가?     |
| **요청 수**        | 적음 (1~수십 건)                   | 많음 (수백~수천 건 이상)        |
| **실패 시 의미**     | 코드 로직 오류                     | 시스템 병목, 성능 한계 노출       |
| **도구/프레임워크** | `unittest`, `pytest`, `APIClient` | `locust`, `gevent`           |

#### 출력 예시

![](https://velog.velcdn.com/images/hak_0/post/779cefe8-f59e-47bc-b9c7-1034f8912eb2/image.png)

브루트포스(Brute Force) 공격을 방어 하기위해 로그인 실패 횟수를 기억하고, 일정 횟수이상 실패하면 일시 차단하는 기능이 필요

django-axes

Django에서 로그인 시도 추적&차단을 해주는 오픈소스 패키지

기본 특징은

Redis + 커트서마이징 방식이 필요한 이유

Django 로그인 시도 제한 방식 비교: django-axes vs Redis + 커스터마이징

💻 커스터 마이징 redis + django 구현 예

import os
import redis
from rest_framework.exceptions import PermissionDenied

if os.getenv("DOCKER_ENV", "false").lower() =="true":
    REDIS_HOST = os.getenv("REDIS_HOST", "redis")
else:
    REDIS_HOST = "localhost"

r = redis.Redis(host=REDIS_HOST, port=6379, db=0, decode_responses=True)

def get_login_attempt_key(key):
    return f"login_attempt_{key}"

def check_login_attempt_key(key, limit=5, block_time = 300):
    redis_key = get_login_attempt_key(key)
    attempts = r.get(redis_key)

    if attempts and int(attempts) >= limit:
        raise PermissionDenied(
            detail="로그인 시도 횟수를 초과했습니다. 잠시후 다시 시도하세요",
            code="Too_much_attempts"
        )
    pipe = r.pipeline()
    pipe.incr(redis_key)
    pipe.expire(redis_key, block_time)
    pipe.execute()

def reset_login_attempt(key):
    redis_key = get_login_attempt_key(key)
    r.delete(redis_key)

Redis pipline() 메서드 정리

• redis.Redis(host="host", port=6379, db = 0, decode=response=True) decode => 기본적으로 UTF-8 로 디코딩해서 사용하기 위해

Login API 적용

class UserLoginView(APIView):
    @swagger_auto_schema(
        security=[{"Bearer": []}],
        request_body=UserLoginSerializer,
        responses={
            200: UserLoginSerializer,
            400: openapi.Response(
                description=(
                    "잘못된 요청 시 응답\n"
                    "- `code`:`missmatch` , 이메일 또는 비밀번호 불일치.\n"
                ),
            ),
            403: openapi.Response(
                description=(
                    "- `code`:`not_verified`, 인증되지 않은 이메일\n"
                    "- `code`:`Too_much_attempts`, 로그인 시도횟수 5회 초과 실패 5분 간 불가"
                )
            ),
        },
    )
    def post(self, request):
        email = request.data.get("email")
        password = request.data.get("password")

        check_login_attempt_key(email)

        user = authenticate(email=email, password=password)
        if not user:
            return Response(
                {
                    "error": "이메일 또는 비밀번호가 올바르지 않습니다.",
                    "code": "missmatch",
                },
                status=status.HTTP_400_BAD_REQUEST,
            )

        if not user.email_verified:
            return Response(
                {
                    "error": "이메일 인증이 완료되지 않았습니다. 이메일을 확인해주세요.",
                    "code": "not_verified",
                },
                status=status.HTTP_403_FORBIDDEN,
            )

        refresh = RefreshToken.for_user(user)
        access_token = str(refresh.access_token)

        store_access_token(user.id, access_token, 3600)
        store_refresh_token(user.id, str(refresh), 86400)

        # activity log 추가 = 로그인
        ActivityLog.objects.create(
            user_id=user,
            action="LOGIN",
            ip_address=get_client_ip(request),
        )

        reset_login_attempt(email)

        return Response(
            {
                "access_token": access_token,
                "refresh_token": str(refresh),
                "token_type": "Bearer",
                "expires_in": 3600,
            },

왜 쓰는가 ?

• API 남용 방지(과도한 요청)
• 비즈니스 정책 반영(무료/유료 티어)
• 서버 자원 보호
• 보안 목적으로는 사용 X(IPsppofing 쉽게 가능함)

Why Use?

• 비인증 사용자에게 낮은 제한, 인증 사용자에게 더 높은 제한 주고 싶을 때
• 특정 API가 리소스를 많이 써서, 해당 API만 따로 제한 걸고 싶을 때
• 짧은 시간(버스트) 과 긴 시간(지속적) 요청 제한을 함께 쓰고 싶을 때

설정 방법

1. settings.py 전역 설정

REST_FRAMEWORK = {
    # 전역설정
    'DEFAULT_THROTTLE_CLASSES' : [
        'rest_framework.throttling.AnonRateThrottle', # 비인증
        'rest_framework.throttling.UserRateThrottle',
    ]
    # 설정 값
    'DEFAULT_THROTTLE_RATES' : {
        'anon': '100/day',
        'user': '1000/day', 
    }
}

• DEFAULT_THROTTLE_CLASSES 를 설정해놓으면 전역에 설정된다.
• DEFAULT_THROTTLE_RATES 를 설정하고 개인 API에 적용시키면 덮어써 진다
• 전역 설정을 하지 않고 API마다 따로 설정하고 싶으면 아래의 RATES 부분만 settings에 작성한 후 개인 API에 throttle_class 를 지정해 주면 된다.

2. 개별 View에서 설정

   from rest_framework.throttling import UserRateThrottle
   

class MyView(APIView): throttle_classes = [UserRateThrottle]

3. @api_view와 데코레이터
```python
@api_view(['GET'])
@throttle_classes([UserRateThrottle])
def my_view(request):
    ...

주요 Throttle 클래스들

사용자 구분 기준

• X-Forwarded-For 헤더 -> 프록시를 거친 경우에 사용됨
  • X-Forwarded-For 헤더는 클라이언트의 원래 IP주소를 알기 위해 사용하는 HTTP 헤더이다. 이 헤더는 프록시나 로드밸런서 같은 중간 서버를 거쳐서 들어온 요청에 추가된다.
  • ex) X-Forwarded-For: 10.0.0.1, 123.123.123.1
    • 123.123.123.1 은 실제 클라이언트 IP
    • 나머지는 프록시나 로드밸런스 IP 이다 장고는 이걸 기준으로 클라이언트 IP를 뽑으려 한다.
• 없으면 Remote_ADDR 사용
  • X-Forwarded-For가 없으면 장고는 기본적으로 WSGI 환경 변수인 REMOTE_ADDR을 사용한다 이건 실제 서버에 도달한 요청의 IP인데, 프록시를 거쳤다면 프록시의 IP 일 수 있어서 정확하지 않을 수 있음
• NUM_PROXIES 설정으로 실제 클라이언트 IP 추출 가능
  • X-Forwarded-For에서 실제 클라이언트 IP를 뽑을 ㄸ'ㅐ, 프록시를 몇개 거치는지 알려주는 설정

    REST_FRAMEWORK = {
    'NUM_PROXIES': 2
    

X-Forwarded-For: 123.123.123.1, 10.0.0.1, 10.0.0.2

앞에 1번째 = 실제 클라이언트 , 나머지 2개 NUM_PROXIES proxy ip 로 보는 것이다.

}

### 💻 예시: Burst + Sustained 이중 제한

* Burst, Sustained
```python
class BurstRateThrottle(UserRateThrottle):
    scope = 'burst'
class SustainedRateThrottle(UserRateThrottle):
    scope = 'sustained'

REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': [
        'my_app.throttles.BurstRateThrottle',
        'my_app.throttles.SustainedRateThrottle',
    ],
    'DEFAULT_THROTTLE_RATES': {
        'burst: '60/min',
        'sustained' : '1000/day',
    }
}

# 이중 제한 적용
class myapp_view(APIView):
    throttle_classes = [BurstRateThrottline, SustainedRateThrottle]

burst(버스트)

• 짧은 시간 동안 허용되는 최대 요청 수
• Ex) 60/min -> 1분에 최대 60번 요청 허용

목적
"잠깐동안 급하게 많은 요청을 보내는 건 허용하되,
너무 오래 그렇게 보내지는 못 하게 하자"

비유
손님이 순간적으로 몰려드는건 괜챃지만,
계속 몰리면 주방이 터지는 거니까 그런 걸 막는 장치

sustained(지속)

• 긴 시간 동안 허용되는 총 요청 수를 의미
• Ex) 1000/day -> 하루에 최대 1000번 요청 가능

목적
"서버 자원을 하루 단위로 적절히 분배해서 남용 방지"

비유
하루에 너무 많이 오는 단골손님을 쫓아낼 수는 없지만
일정 이상은 못 사게 제한하는 것

burst + sustained

• 사용자가 1분 안에 60번 넘게 요청 -> burst 초과로 차단
• 사용자가 하루 종일 천천히 요청해서 1000번 도달 -> sustained 초과로 차단

결론

Throttle를 사용하는데 하나의 제한이 아닌 burst + sustained로 나눠서 두개의 제한을 둘수 있다.
또한 인증된 User와 그렇지 않은 유저의 제한을 나눌수 있다.
UserRateThrottle -> 인증된 유저
AnonRateThrottle -> 인증되지 않은 유저

각각을 상속받아 스코프를 나누어 제한을 나눌수도 있다.

API 프로젝트를 하다 Throttle에 대한 개념을 정리해 보았다.

sqlmigrate

• makemigartions 를 하면 DB에 실행 되기전의 파일이 생성된다.

• sqlmigrate는 마이그레이션이 실제로 어떤 SQL을 실행할지 미리 확인하는 명령어 이다.
• 데이터베이스에 변경을 적용하지 않고, SQL만 출력한다.
• python manage.py sqlmigrate <앱이름> <마이그레이션 번호>

Model

• auto_now: 세이브 할때마다 수정
• auto_now_add: 처음 넣을때 생성

Django Settings

• 시간설정은 일단 우선 한국기준으로 세팅하자

  # shettings.py
  LANGUAGE_CODE = "en-us"
  

TIME_ZONE = "Asia/Seoul"

USE_I18N = True

USE_TZ = False

### 단위 테스트로 잡을수 없는 동시성 문제

#### test
> * `poetry add django-ninja` 
>   - DRF 보다 가볍고 FastAPI 스타일을 지원하는 장고 라이브러리 스웨거 문서지원

```python
from ninja import NinjaAPI
api = NinjaAPI()


@api.get("like")
def api_do_like(request: HttpRequest, user_id: int, article_id: int) -> None:
    do_like(user_id, article_id)
    return None

• swagger-ui 를 사용해 데이터를 넣어 본다 그리고 http url를 찾는다

• ctx.captured_queries : Django의 테스트 환경에서 실행된 데이터베이스 쿼리를 캡처할 때 사용하는 속성

  from django.test.utils import CaptureQueriesContext
  from django.db import connection
  from myapp.models import User
  

with CaptureQueriesContext(connection) as ctx: User.objects.filter(is_active=True).count() # 실행할 쿼리

실행된 쿼리 확인

for query in ctx.captured_queries: print(query["sql"]) # 실행된 SQL 출력

#### httpie 를 사용해 다발성으로 넣어본다
> * curl 대신 http를 사용할수 있게 해주는 패키지
> * `brew install httpie`

```HTML
http ":8000/api/like?user_id=2&article_id=2" &
http ":8000/api/like?user_id=2&article_id=2" &
http ":8000/api/like?user_id=2&article_id=2" &
http ":8000/api/like?user_id=2&article_id=2" &
http ":8000/api/like?user_id=2&article_id=2" &

이 코드를 터미널에 실행시키면 DB에서 한개가 아닌 여러개의 데이터가 들어간 것을 확인할 수 있다.

• View코드 에서 같은 코드가 들어가는걸 막아놨다 문제는 거의 동시에 같은 입력값을 넣으면 DB에서 데이터가 없다고 판단하고 넣어버리는 것이다.

이 문제를 해결하기 위해선 Model의 내용을 수정해서 그전에 차단해 주어야한다.

UniqueConstarnt

class Like(BaseModel):
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    article = models.ForeignKey(Article, on_delete=models.CASCADE)
    class Meta:
        constraints = [
            models.UniqueConstraint(fields=["user","article"], name="UIX_user_id_article_id")
        ]

이후 migrate 를 진행한 후 다시한번 넣어보면 하나의 컬럼만 형성된 것을 확인할 수 있다.

Debug

테스트 코드에서 중단점을 찍어 디버그를 할수 있다. 디버그시 변수명, 타입명 등을 바로 찾을수 있고 어떤 에러가 발생했는지 확인할 수도 있다. Exception 으로 처리해놓은 에러 확인가능.

중단점을 찍어야 하는 주요 위치

✅ 함수의 시작 부분 → 입력 값이 올바른지 확인 ✅ 조건문 내부 (If-Else) → 어떤 분기로 가는지 확인 ✅ 반복문 내부 (For / While) → 루프가 정상적으로 도는지 확인 ✅ 예외가 발생할 가능성이 있는 부분 → 오류 원인 분석 ✅ 변수 값이 변경되는 부분 → 데이터 흐름 추적

Evalutae

PyCharm의 Evaluate Expression (표현식 평가) 기능은 디버깅 중 특정 변수의 값을 확인하거나 표현식을 실행할 때 사용하는 도구입니다.

📌 주요 기능 ✅ 디버깅 중 특정 변수 값 확인 ✅ 특정 코드나 표현식을 실행하여 결과 확인 ✅ 새로운 값으로 변수 변경 가능

WAS(Web Apllication Server)

웹 서버와(HTTP요청 처리)와 애플리케이션 실행 환경을 제공하는 서버

사용자의 요청을 처리하고 DB와 연동하며 동적인 응답을 생성

웹서버는 정적 콘텐츠 제공, WAS는 동적인 로직 처리를 담당.

WAS는 웹 애플리케이션이 실행되는 핵심 서버로, 클라이언트의 요청을 받아서 비즈니스 로직을 실행하고 응답을 생성하는 역할을 합니다. 웹 서버와 함께 사용되며, 다양한 기술과 연계하여 확장성과 보안성을 높일 수 있음

settings.py

Author.objects.using("default") # default 로 정의해 놓은 DB를 사용 Author.objects.using("read") # 읽기 전용으로 지정해 놓은 DB사용

coverage

• 테스트가 코드의 몇 퍼센트를 실행했는지 측정하는 지표

• poetry run coverage run manage.py test 테스트 실행 및 커버리지 측정

• poetry run coverage report -m 커버리지 리포트 출력

• miss -> 1줄: 한번도 실행되지 않은 줄 체크

• -m => show missing : 어느 라인에서 버그가 발생했는지 확인

coverage 는 전체코드 분의 제품 테스트 코드 이다. 그래서 모든 함수에 대해 테스트코드를 작성해야한다.

[tool.coverage.report]
fail_under = 80 # 몇프로를 기준으로 coverage가 됬는지 하는 것

[tool.coverage.run]
omit = [ # coverage 측정을 할 필요가 없는 것들
  ".mypy_cache/**/*", # mypy가 내부적으로 하는 커버리지
  "*/migrations/*", # 장고에서 만든건데 할 필요가 없음
  "*/*test*.py", # 테스트 코드 할필요 없음
  # 앱의 실제 코드가 테스트에 의해 얼마나 실행되었는지 측정하는 것인데 
  # 테스트를 수행하는 코드이기에 측정할 필요가 없음
]

ci 파일에 이걸 넣는다

      - name: Test python project
        run: |
          poetry run coverage run manage.py test
          poetry run coverage report -m

• 리눅스에서 성공 여부를 체크할 수 있다. echo $? 바로 직전의 코드를 출력하는 명령어인데 eixt code 가 출력된다 리눅스에서 exit 코드는 0을 제외한 다른 숫자는 전부 오류가 발생했단 것이다

용어

• LazyLoad : 정말로 필요하기 전까지 가져오지 않는것을 LazyLoad라 한다.
• eagerload <=> prefetch : 미리 가져오는것

select related : join prefetch related : 개별 쿼리가 생성

RDBMS

가장큰 특징 : 트랜잭션

서버가 다운되면 앞에 내용이 처리되는 것을 방지하기위해 전부 실패되거나 전부 성공을 처리한다 이것이 Atomicity 이다.

기본 구조는 모든것을 실행 후 commit or Rollback 을 하는 것이다.

기본 구조는 원래 BEGIN; 을 시작한후 모든 처리동작을 한후 COMMIT 을 하는것이다 or Rollback

쉽게말해 다른 브랜치로 이동하여 작업하는 것이다 그래서 다른곳에서의 일이 영향을 받지 않는다.

ACID

• A, Atomicity: 일련의 데이터베이스 연산이 모두 성공하거나 모두 실패하는 특성. (all or nothing)

• C, Consistency: 트랜젝션이 끝나도 integrity 가 유지되는 특성. (foreign key 가 없는 id 를 가리키거나 하지 않습니다.)

• I, Isolation: 한 트랜젝션에서 일어난 변경사항이 다른 트랜젝션에서 조회되는 정도를 의미합니다. mysql 에서는 다음과 같은 isolation level 이 존재합니다.(innodb 기준) https://dev.mysql.com/doc/refman/8.0/en/innodb-transaction-isolation-levels.html

  • SERIALIZABLE
  • REPEATABLE READ (디폴트) (다른 트랜젝션에서의 변경사항은 commit 되기 전까지는 안 보인다)
  • READ COMMITTED
  • READ UNCOMMITTED

• D, Durability: 트랜젝션이 COMMIT 되고나면 영구적으로 저장되는 특성을 의미합니다.

• sifht 2번 파이참 내부의 모든걸 찾는 단축키
• cmd + option + k : 커밋과 푸쉬를 한번에 하는 단축키
• cmd 엔터 : 커밋
• iterm, poetry 가상환경 사용 : poetry env activate 해서 나온 가상환경 실행 동작 복붙
• shift 2번 모든걸 찾을수 있는 단축키
• cmd + option + k : 커밋과 푸쉬 한번에 하는 단축키
• cmd + shift + n : 아무 파일이나 아무경로에 빠르게 만드는 단축키
• cmd + option + l : 코드 정렬 단축키 -> sql
• cmd + 7 : 구조 확인 데이터베이스
• control + shift + R : 해당 함수만 테스트 실행
• control + shift + D : 해당 함수만 디버깅
• option + enter : 객체 추가 클래스에서
• control + r: 터미널 창에서 명령창 그전에 했떤거 검색가능
• option + 방향키 : 단어 단위로 옮겨감
• fn + 방향키 : 끝과 끝
• 파이참 기능 커맨드+R : 찾기 + 수정
• 내가설정해 놓은것 컨트롤 + a,z : 위아래로 해당영역 설정
• 거기서 움직이려면 움직이면 다같이 움직임 선택하려면 쉬프트 누르고 오른쪽

• HTTP Request Header:
  • Host : 누구에게 메세지를 보낼 것이냐, 받는 사람이름의 역할
  • User-Agent : 보내는 내가 누구냐
  • Content-type : 타입

정적 동적

static : 정적, 실행 전에 결정 dynamic : 동적, 실행 중에 결정

Git + 파이참

• git init : 왼쪽 commit 관리하는 확장파일 생성됨
• 커밋전 무엇을 커밋하는지 항상 확인하고 실행하는 습관
• I/O 동작이 일어나는 것은 필요한지 안필요한지 항상 체크 시스템에 큰 부하를 일으킴 ex) print

DB 연결

파이참 오른쪽 db 에서 연결 시키기 안에서 sql문 실행 시켜서 확인

poetry add pymysql cryptography

poetry add -D **types-PyMySQL**

pymysql.install_as_MySQLdb()
DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.mysql",
        "NAME": "oz_django",
        "USER": "root",
        "PASSWORD": "1234",
        "HOST": "localhost",
        "PORT": "3306",
    }
}

개발시 테스트 환경 구축 및 통일

• poetry add django-stubs : 장고의 타입힌팅 지원 패키지가 설치

  • 정적 타입 분석을 위한 패키지
  • mypy와 함께 Django 코드의 타입 검사를 수행
  • Python의 기본적인 타입 힌팅 뿐만 아니라, Django ORM 관련 타입도 제공

• dev dependency <--> dependency (poetry add -D 패키지, poetry add 패키지)

  • dev : 개발환경에서만 사용하는 패키지
  • 기본: 실제 배포환경에서의 패키지
  • 개발하다가 이건 개발환경에서만 쓰겟다 싶은건 -d로해서 추가로 깔고 기존에 있떤건 삭제시킨다

• 프로젝트 생성

• ignore 생성, .gitignore

  .idea/
  venv/
  __pycache__/

ipython

`poetry add ipython

black

• codeformatter: 여러 개발자의 코드 스타일을 통일시켜주는 패키지
  • ex) ' ' -> " " , 공백... 맞춰진것 commit 목록에서 확인 가능
• poetry add -D black

• poetry run black .

• poetry.toml 추가

  # 자동으로 넘겨지는것 방지 기존이 80
  [tool.black]
  line-length = 120
  

파이참 수정

preference -> editor -> code style -> hard wrap at : 설정한 블랙 값으로 수정

### isort
> * `poetry run isort .`
> * 알파벳 순서를 맞추는 정도의 동작. 

* poetry.toml 추가
```toml
[tool.isort]
profile = "black"

Mypy

• 코드를 읽고 잠재적인 에러를 찾는다
• poetry run mypy .

• poetry.toml 추가

  [tool.mypy]
  plugins = ["mypy_django_plugin.main"]
  python_version = "3.13"
  strict = true # mypy 진가 타입을 엄격히 써야함
  

[[tool.mypy.overrides]] # 특정한 모듈에 한해서 덮어쓰겠다. module = ".migrations." ignore_errors = true # 위의 폴더에 관해서는 mypy를 무시하겠다. 어차피 장고에서 만들어주니

[[tool.mypy.overrides]] module = "manage" ignore_errors = true # 위의 폴더에 관해서는 mypy를 무시하겠다. 어차피 장고에서 만들어주니

[tool.django-stubs] django_settings_module = "폴더명.settings" # settings의 경로 명시

* mypy 에서는 튜플, 리스트, 딕 이런건 의미 없기에 None 빈 형태는 의미가 없다 그래서 내부의 요소에도 type을 정해줘야 한다. => mypy strict True 옵션
* `reveal_type(어떤 타입이 들어가 있는지 확인할 타입명)` : from typing import reveal_type
* mypy는 정적이다, 그리고 코드를 실행해 보지 않는다. 

### Github Action
> * Continuous integration(CI)
> * github에 push 할 때마다 테스트가 자동으로 실행되도록 만들기
> * ci.yml 안에있는 내용을 github에서 docker처럼 새로 만들고 테스트 해본후 없앤다.
> * 독립된 환경에서 독립적인 자기만의 데이터를 가지고 실행

- Continuous IntegrationCI 가 필요한 이유 : 수십 명이 하나의 코드베이스에서 작업을 할 때 
다음과 같은 과정을 반복하게 됩니다.

    - 테스트가 끝난 코드(master)에서 새로운 branch 를 생성합니다.
    - 새로운 branch 에서 작업합니다. (기능 추가, 최적화 등등)
    - 로컬에서 테스트 합니다.
    - PR 을 올립니다. 코드리뷰, 테스트
    - 개발서버에 배포 합니다. (staging), 개발 서버에서 기계를 통해 기존의 기능까지 전부 잘되나 검증
    - develop 서버에서 테스트 합니다.
    - master 에 merge, 배포합니다.


 이렇게 작업할 때 마다 반복적인 작업이 있습니다. PR 을 올리기 전에 테스트를 다 통과해야 한다는 점, merge 할 때 마다 테스트 해야 하는 점 등이죠.

 이를 자동화 함으로써

- human error 를 줄입니다.
- 반복적인 작업에 들어가는 시간을 아낌으로써 더 효율적으로 일할 수 있습니다.

* 아래의 내용을 커밋해서 push 합니다.
* 깃허브 액션 스크립트, `.github/workflows/ci.yml` 폴더 및 파일 생성
```yaml
name: Django CI

# 트리거 언제 실행한건 지
on:
  push: # 푸쉬 할때마다
#  workflow_dispatch: # 그 버튼을 누르면 실행

jobs:
  ci:
    runs-on: ubuntu-22.04 # github action 이 제공하는 ubuntu는 mysql 이 기본적으로 설치되어 있음
    env:
      # 환경 변수
      DB_HOST: 127.0.0.1
      DB_PORT: 3306
      DB_USER: root
      DB_PASSWORD: 1234 # 깃허브에 넣는거라 비밀번호 의미없음
      DB_DATABASE: oz_django

    steps: # 단계
      - name: Check out the codes
        uses: actions/checkout@v2

      - name: Setup python environment
        id: setup-python
        uses: actions/setup-python@v2
        with:
           python-version: '3.13'

      - name: Set timezone to KST
        run: |
          sudo rm /etc/localtime
          sudo ln -s /usr/share/zoneinfo/Asia/Seoul /etc/localtime

      # Start Mysql
      # https://ovirium.com/blog/how-to-make-mysql-work-in-your-github-actions/
      - name: Start Mysql
        # 이미 설치된 mysql을 단순히 킴
        run: |
          sudo systemctl start mysql
          mysql -e "use mysql; FLUSH PRIVILEGES; ALTER USER '${{ env.DB_USER }}'@'localhost' IDENTIFIED BY '${{ env.DB_PASSWORD }}';" -uroot -proot
          mysql -e 'CREATE DATABASE ${{ env.DB_DATABASE }};' -u${{ env.DB_USER }} -p${{ env.DB_PASSWORD }}
      - name: Install Poetry
        run: |
          curl -sSL curl -sSL https://install.python-poetry.org | POETRY_VERSION=2.0.1 python3 -
          echo "${HOME}/.local/bin" >> $GITHUB_PATH

      - name: Install dependencies
        run: |
          poetry install --no-root

      - name: Run black
        # check 만 하고 실제론 바꾸지 않는다 우리 프로젝트에서 바꿔야 하니까
        run: |
          poetry run black . --check          

      - name: Run isort
        # 마찬가지
        run: |
          poetry run isort . --check --diff

      - name: Run Mypy
        run: |
          poetry run mypy .

#      - name: Test python project
#        run: |
#          poetry run python manage.py test

• on: push → 액션이 언제 실행되는지를 정의합니다. on: push 라면, push 할 때마다 실행합니다. https://docs.github.com/en/actions/learn-github-actions/workflow-syntax-for-github-actions#on
• jobs→ 하나의 workflow 는 여러개의 job 으로 구성됩니다. 여기서는 ci 라는 이름(id)의 job 하나만 정의하여서 사용하였습니다. https://docs.github.com/en/actions/learn-github-actions/workflow-syntax-for-github-actions#jobs
• runs-on → job 이 실행되는 machine 을 의미합니다. github 가 제공하는 ubuntu-latest (현재 버전 20) 을 사용합니다. https://docs.github.com/en/actions/learn-github-actions/workflow-syntax-for-github-actions#jobsjob_idruns-on
• steps → 하나의 job 은 여러개의 step 으로 구성됩니다. step 은 명령을 실행하거나 다른 action 을 실행합니다.
• uses → 실행할 action 을 가리킵니다.
• with → action 에 전달할 parameter 변수입니다.
• run → 실행할 명령어입니다.
• run: | → yaml 문법입니다. | (파이프라인)을 사용해 value 가 여러 줄(multiline) 이라는 것을 의미합니다.

TL; DR:

push 될 때 마다 → 우분투에서 → 파이썬을 설치하고 → poetry, django, 등등등 종속성 설치하고 → test 를 실행한다.

• push

  git remote add origin https://github.com/deokbaae/oz_django.git

  • pycharm 에서 shift 2번 연타 후 push

    github login

• 매번 테스트 코드를 사용할 수 없으니 test.sh 제작 poetry add coverage 패키지 설치를 한다.

• chmod +x ./test.xh 를 통해 관리자 권한으로 허용해준다

• ./test.sh

  set -eo pipefail
  

COLOR_GREEN=tput setaf 2; COLOR_NC=tput sgr0; # No Color

echo "Starting black" poetry run black . echo "OK"

echo "Starting isort" poetry run isort . echo "OK"

echo "Starting mypy" poetry run mypy . echo "OK"

echo "Starting test with coverage" poetry run coverage run manage.py test poetry run coverage report -m

echo "${COLOR_GREEN}All tests passed successfully!${COLOR_NC}"

```

Agile (애자일) 개념 정리

📝 애자일(Agile) 개요

Agile(애자일) 은 소프트웨어 개발 및 프로젝트 관리에서 유연하고, 빠르고, 협업 중심으로 작업하는 방법론입니다.
기본 철학: 작은 단위로 빠르게 개발하고, 지속적으로 피드백을 반영하여 개선하는 방식

애자일의 핵심 원칙

애자일의 기본 원칙은 "애자일 선언(Agile Manifesto)"에서 정의되었습니다.

1. 프로세스보다 개인과 팀워크를 중시
2. 방대한 문서보다 실행 가능한 소프트웨어를 중시
3. 계약 협상보다 고객과의 협력을 중시
4. 초기 계획보다 변화에 대한 유연성을 중시

즉, 빠른 피드백, 반복적인 개선, 고객과의 협업이 애자일의 핵심입니다. 🚀

애자일의 대표적인 방법론

애자일 프로세스 (Scrum 예시)

1. 백로그(Backlog) 작성 - 해야 할 작업 목록 정의
2. 스프린트 계획(Sprint Planning) - 1~4주 동안 개발할 목표 설정
3. 데일리 스탠드업(Daily Stand-up) - 매일 15분 회의로 진행 상황 공유
4. 스프린트 개발(Sprint Execution) - 개발 진행 (코딩, 테스트, 배포)
5. 스프린트 리뷰(Sprint Review) - 고객과 함께 기능 시연 및 피드백
6. 회고(Retrospective) - 개선할 점 논의 후 다음 스프린트 반영

📌 이 과정이 반복되면서 지속적인 개선이 이루어짐!

애자일 vs 전통적 개발 (워터폴)

🔥 애자일은 유연하게 변화에 대응 가능, ⚠️ 워터폴은 계획이 확실할 때 적합

애자일 개발의 장점

✅ 빠른 피드백 반영 → 고객 요구사항을 신속하게 적용 가능
✅ 변화에 유연하게 대응 → 요구사항 변경 시 큰 문제 없이 진행 가능
✅ 지속적 개선 & 품질 향상 → 짧은 주기로 반복 개발하며 품질 향상

최종 요약

DTO (Data Transfer Object)란?

DTO(Data Transfer Object) 는 데이터를 전달하기 위한 객체로, 클라이언트와 서버 또는 애플리케이션 내부에서 데이터를 효율적으로 주고받기 위해 사용됩니다.

DTO의 주요 역할

1.    데이터 전달을 위한 객체 (Request / Response)
•    API 요청/응답 시 데이터를 정리하여 전달
2.    불필요한 데이터 노출 방지
•    민감한 정보를 숨기고 필요한 데이터만 포함 가능
3.    데이터 변환 (예: 모델 → DTO, DTO → JSON)
•    ORM 모델을 API 응답 형식으로 변환할 때 사용

📌 쉽게 말하면? ➡️ “딕셔너리(dict)보다 구조화된 데이터 전송을 위해 사용하는 클래스!”

• Operating System
• ex) windows, android, linux
• 운영체제는 메모리를 이용해서 새로운 동작을 하고 그안에서 어떤 데이터를 다룰수 있게 인터페이스를 제공해주는 역할을 함

• 운영체제는 플랫폼
  • 운영체제란, 컴퓨터에 포함된 CPU나 메모리, 입출력 기기 등이 사용자에 기대에 맞게 역할을 수행할 수 있도록 도와주는 창구 역할을 하는 소프트웨어
  • 이러한 역할을 플랫폼 소프트웨어라고 부르기도 함
  • 사용기기나 목적에 따라 필요한 운영체제의 유형이 다르기 때문에 운영체제의 종류가 다양함

📝 운영체제의 대표적인 역할들

• 프로세스 관리
• 메모리 관리 : 데이터 들을 필요한 공간에 맞게 사용
• 파일 시스템 관리

📝 운영체제의 구조

• 커널

• 커널: 사용자가 직접 접근 X, 아주 핵심적이고 중요한 파트이기에
  • 사용자가 커널에 요청을 CLI(Command Line Interface)를 통해 할 수 있음
  • 또는 GUI(Grapic User Interface) -> 데스크톱 컴퓨터를 키면 바탕화면 같은것이 GUI임
• 시스템 콜
  • 어플리케이션이 컴퓨터의 자원에 접근할 떄 위험성 없이 적절한 방법으로 작업을 수행 할 수 있게 해주는 어플리케이션과 커널간의 인터페이스 이다.
  • 쉽게 말해서 커널에서 정해준 함수만 사용할 수 있음
• 드라이버
  • 그래픽카드, USB, 메로리 등 다양한 외부기기와 컴퓨터를 연결했을때 정상적으로 사용되려면 운영체제가 지원을 해줘야 사용을 가능

📝 컴퓨터 하드웨어 구조

CPU(Central Processing Unit): 중앙 처리 장치

• 컴퓨터 내부에서 주요 연산을 처리하느 두뇌 역할
• ALU(Arithmetic Logic Unit, 산술 논리 연산 장치): 연산 담당 파트
• CU(Control Unit): 연산 명령을 해석
• Register Set: 데이터를 처리하기 위해 임시 저장 장치

Main Memory(Radom Access Memory : RAM)

• 프로그램의 실행 파일이 올라가서 실행되는 영역
• 프로세스 관리를 위해 존재하는 메모리

버스 시스템

컴퓨터를 구성하는 요소들 사이에서 데이터를 주고받기 위해 사용되는 경로

1. 데이터 버스(Data Bus)

   • 역할: CPU와 메모리, 입출력 장치 간 데이터를 전달
   • 특징: 양방향 전송 가능
   • 예시: CPU가 RAM에서 데이터를 읽거나 저장할 때 사용

2. 주소 버스(Address Bus)

   • 역할: CPU가 메모리나 입출력 장치의 특정 주소를 지정
   • 특징: 단방향 전송 (CPU → 메모리/입출력 장치)
   • 예시: CPU가 특정 메모리 주소에 접근할 때 사용

3. 제어 버스(Control Bus)

   • 역할: CPU가 시스템의 다양한 장치들을 제어하는 신호를 전달
   • 특징: 양방향 또는 단방향 전송
   • 예시: 읽기(Read), 쓰기(Write), 인터럽트(Interrupt) 신호 등

📌 한마디로 정리!

• 데이터 버스: 데이터 이동
• 주소 버스: 위치 지정
• 제어 버스: 명령 전달

📝 CPU 이해

• ALU(Arithmetic and Logic Unit, 산술 논리 연산 장치): 연산 담당 파트
  • 산술 연산
  • 논리 연산
  • 바이너리 형태를 이해하는 능력이 없음 그래서 CPU에 등록된 명령어를 해석하기 위해 CU가 필요함
• CU(Control Unit): 연산 명령을 해석, C나 JAVA같은 compile 언어로 작성한 코드를 compile하면 실행 파일이 만들어짐, 실행 파일엔 CPU에 일을 시키기 위한 명령어가 저장되어 있음
• Register Set: 데이터를 처리하기 위해 임시 저장 장치
  • 프로그램 카운터(Program Counter): 메모리에서 가져다 실행할 다음 명령을 일시적으로 저장하는 레지스터
  • 인스터럭션 레지스터(Instruction Register): 이제 해석해야할 명령어를 저장하는 레지스터
  • 어드레스 레지스터(Address Register): 메모리의 주소를 저장하는 레지스터, CPU가 읽어들이고자 하는 주소값을 어드레스 버스(Address Bus) 로 보낼때 거치는 레지스트
  • 버퍼 레지스터(Buffer Register): 메모리의 읽거나 쓰려는 데이터나 명령을 일시적으로 저장하는 레지스터, 데이터를 주고받는 데이터가 바로 쏴지지않고 잠시 대기하는 상태
  • 플래그 레지스터(Flag Register): 연산 결과, CPU상태에 대한 부가적인 정보를 저장하는 레지스터
  • 스택 포인터(Stack Pointer): 스택의 꼭대기를 가리키는 레지스터, 스택이란 지역변수나 매개변수를 저장하기위해 사용하는 메모리 영역의 이름, 다음꺼내 써야할 데이터를 알려주기위해 사용하는 레지스터
  • 플래그

1. CPU 명령전달
2. CU 명령 해석
3. ALU 명령 실행
4. 일을 하는 와중 중 명령이 들어오면 Register에서 대기

• 클럭 펄스(Clock Pulse)

  • 시스템 동기화를 위해 존재하는 클럭 펄스
  • 초당 처리할 수 있는 명령의 갯수, CPU는 그 클럭에 맞춰서 일을 진행함
  • 동기 명령을 가지런히 정렬해서 처리하기 위해 클럭에 맞춰서 하는것

• 즉 이러한 시스템을 매끄럽게 진행되도록 도와주는 것이 운영체제 이다.

📝 프로그램 실행 과정

폰 노이만 구조

CPU <--> 메모리 <- 프로그램

• 프로그램 코드(인간친화적) -> 어셈블리 코드(컴퓨터 친화적) -> 바이너리 코드(1과 0으로 구성된 코드)

1. 실행파일이 만들어짐 -> 하드디스크 같은 보조메모리에 저장
2. 메인 메모리에 등록
3. CPU에 의해 차레대로 실행(순차적, 스택으로 쌓여있음)
   1. Fetch: 메모리 상에 등록된 명렁어를 CPU로 가져옴(레지스트)
   2. Decode: 가져다 놓은 명령어를 CPU가 해석(CU)
   3. Execution: 수행(ALU)

버스 시스템

• 데이터 버스: 데이터 이동을 위해 필요한 버스
• 컨트롤 버스: CPU가 원하는 바를 메모리에 전달하기 위한 버스
• 어드레스 버스: 주소값을 이동하기 위해 필요한 버스

📝 프로그램 실행 중 돌발상황, 인터럽트

• CPU가 어떤 작업을 수행하고 있을 때
• CPU의 작업을 방해하는 신호를 가리켜 인터럽트
• -> CPU가 지금 하고있는 작업을 잠시 멈추고 우선적으로 처리해야할 작업이 생겼다라고 신호를 보내는 것

• 정상적으로 수행할 수 없는 명령어가 입력되면, CPU는 인터럽트를 발생시킨다 -> 예외 Ex) ValueError, ZeroDivisionError...
• 입출력 장치(하드웨어)로부터 발생하는 인터럽트는 비동기 인터럽트라고 표현 하기도 함 CPU가 발생시키는 인터럽트가 아니기에 Ex) KeyBoardInterrupt(ctrl+c) ...

인터럽트 핸들링

• 인터럽트가 발생하면, 인터럽트에 대응하는 무언가가 동작해야 한다. 이를 인터럽트 서비스 루틴 이라한다.
• 인터럽트 서비스 루틴은 이터럽트를 처리하기 위해 특정 인터럽트 신호에 대해 미리 정의되어 있는 프로그 또는 함수 이다.

import time
import signal

def handler(signum, frame):
    print("키보드 인터럽트 감지")
    print("신호 번호: ", signum )
    print("스텍 프레임:", frame)
    exit()

signal.signal(signal.SIGINT, hanlder)

while True:
    print("5초 간격으로 출력중...")
    time.sleep(5)

📝 프로세스

• 프로세스란 실행중인 프로그램을 뜻함. 운영체제는 필요한 자원을 할당하고 관리, 그렇게 시작한 프로그램은 프로세스 가 됨

프로세스 구조

• 명령어가 담긴 코드 영역
  • 프로세스가 실행하는 코드 즉 명령어 집합이 저장되어 있는 영역
• 전역변수 등이 담긴 데이터 영역
• 지역변수 등이 담긴 스택 영역
  • 휘발성 변수가 저장되어 있는 곳
• 동적 메모리 할당을 위한 힙 영역

레지스터

• 프로그램의 실행을 위해서는 절대적으로 레지스터가 필요
  • 현재 실행중인 프로세스를 위한 데이터들로 채워짐

# 파이썬 프로그램도 프로세스가 될수 있음
import os

# getpid: 운영체제 관리를 위한 프로세스 ID
# 프로세스에는 ID가 할당됨
print("파이썬 코드 실행 중! 실행 중인 프로세스의 아이디는 :", os.getpid())

# 내 컴퓨터 운영체제 안에서 돌아가고 있는 프로세스를 조회하는 코드
# pip install psutil

import psutil

for proc in psutil.process.iter():
    ps_name = proc.name()
    print(ps_name)
    if "Chrome" in ps_name :
        print(ps_name,  proc.pid)

📝 프로세스 상태

• CPU는 한 번에 하나의 연산밖에 수행할 수 있다.

• CPU는 여러 개의 프로세스를 동시에 실행하지 않고, 빠르게 번갈아 가며 실행한다.

• 운영체제는 빠르게 번갈아 수행되는 프로세스의 순서를 관리하고 자원을 분배해야함

  • PCB(Process Control Block): 프로세스와 관련된 자료구조를 저장하는 구조
  • 프로세스를 식별하기 위해 필요한 정보들이 저장됨
  • 프로세스 ID, 레지스터 데이터, 스케줄링 정보, 그리고 상태 ...

• 운영체제는 CPU가 여러개의 프로세스를 번갈아 수행할수 있게 해야함, 그래서 각 프로세스는 각 상황에 따라 상태정보를 가짐

• 생성 -> 준비 <--> 실행 -> 대기 -> 종료, 또는 다시 준비 -> 실행 -> 대기 -> ...종료(소멸)

  • 대기상태는 대기상태가 된 이유가 사라지고 다시 준비상태가 되어야 다시 실행될 수 있음

프로세스 계층

• 부모 프로세스 -> 자식 프로세스 (부모 프로세스에 의해 생성된 프로세스): 독릭적이며 자신만의 ID를 가지고 있음

# 계층을 확인하는 코드
import psutil

for proc in psutil.process_iter():
    ps_name = proc.name()

    if "Chrome" in ps_name :
        child = proc.children()
        print(ps_name, proc.status(), proc.parent(), child)

        if child :
            print(f"{ps_name}의 자식 프로세스", child)

📝 컨텍스트 스위칭

멀티 프로세스 운영체제

• 여러개의 프로세스를 관리해가며, 마치 동시에 실행되는 것처럼 운영할 수 있는 운영체제
• 운영체제가 실행 중일때 다른 프로세스를 여러개를 실행한다는 의미임
• 운영체제에 매우 부담되는 작업

컨텍스트 스위칭

• 프로세스를 실행하는 중에 다른 프로세스를 실행 하기 위해 실행중인 프로세스의 상태를 저장하고 다른 프로세스의 데이터 상태로 교체하는 것을 컨텍스트 스위칭 이라함
• 문맥 교환

• 현재 실행중인 상태의 레지스터를 다 저장해서 프로세스로 넣어 준비상태의 프로세스를 만들고 백업 해둠
• 그리고 실행중인 상태의 프로세스와 번갈아 가면서 사용하는 것임

프로세스 생성

• 프로그램 실행 시, 운영체제는 코드 영역과 데이터 영역을 메인 메모리에 올리고 빈 스택과 빈 힙을 만들어 공간을 확보한다. 이는 시스템에게 상당한 부담을 주는 일
  • 새로운 프로세슽를 생성하는 것보다, 기존 프로세스를 복사하는 것이 빠름, 따라서 모든 프로세스는 최초의 프로세스로부터 복사된다 -> 부모 프로세스 로부터 자식 프로세스가 생성
  • 부모 프로세스를 복사하는 시스템 함수를 fork() 함수라 함
  • fork()를 통해 만들어진 함수를 exec() 시스템 함수를 이용해 코드 영역과 데이터 영역을 원하는 내용으로 덮어쓰기 위한 작업을 해줘야함
  • 그러면 부모 프로세스와 다른 프로세스가 만들어지고 동작하게 됨

# 멀티 프로세싱을 위한 작업

from multiprocessing import Process
import os

def func():
    print("안녕, 함수야!")
    print("나의 프로세스 아이디: ", os.getpid())
    pirnt("나의 부모 프로세스 아이디:", os.getppid()) # 부모 프로세스

if __name__ == "__main__" :
    print("앱 프로세스 아이디 : ", os.getpid())
    child1 = Process(target=func).start()
    child1 = Process(target=func).start()
    child1 = Process(target=func).start()
# 부모 프로세스는 전부 같은 값이 나올것이고 자식은 다르레 나옴
# 만약 함수를 여러개 만들어서 다른것을 출력하더라도 아이디 값은 다르게 나오겟지만
# 한 모듈에서 파생되어서 나오는 것이니 부모 프로세스는 동일하게 나온다

📝 스레드의 이해

• 컨텍스트 스위칭으로 인한 부하를 줄이기 위한 답은 프로세스를 줄이는 것이다.
• 프로세스는 슬하에 스레드를 두어 작업을 처리할 수 있다.

• 모든영역을 공유하지만 스택 영역많은 공유하지 않는다.(지역변수, 매개변수를 저장하는 영역)
  • 작업마다 따로 가지지않으면 제대로 작업이 처리되지 않음 -> 개별적 쓰레드 할당
  • 스택은 함수 호출시 전달되는 인자, 되돌아갈 주소값, 및 지역변수등을 저장하는 메모리 함수이다. 그 구조조차 스택의 구조이다, 따라서 메모리 공간이 독립적이지 않으면 추가적인 실행 흐름을 만들수 없음
• 처리할 작업이 많아져도 프로세스를 늘리는 것이 아닌 스레드를 늘려 효율적으로 처리 가능
• 스레드도 개별적으로 구분이 되어야 하기에 개별 ID가 존재

메모리 구조 관점에서 본 스레드 특징

• 스레드는 스레드가 하나 생성될 때마다 스레드를 위한 스택영역이 추가로 생성될 뿐, 그 이외의 영역은 프로세스 영역을 공유
• 스레드는 프로세스가 처리해야 할 작업을 수행하기 위해 존재하는 것이므로, 코드 영역을 공유해 명령어에 접근할 수 있어야 한다.
• 명령어 실행 시 전역변수, 정적변수, 지역변수 등의 데이터에 접근해야 하므로 데이터 영역과 힙 영역과 공유해야 한다.

# 하위 쓰레드 만들기
import threading
import os

def func():
    print("안녕, 함수야!")
    print("나의 프로세스 아이디: ", os.getpid())
    print("스레드 아이디:", threading.get_native_id())

if __name__=="__main__" :
    print("기존 프로세스 아이디 :", os.getpid())
    thread1 = threading.Thread(target=func) # target에 맡길 작업 투척
    thread1.start()

---
import threading
import os
import time

def something(word):
    while True:
        print(word)
        time.sleep(3)

if __name__=="__main__" :
    print("기존 프로세스 아이디 :", os.getpid())
    t = threading.Thread(target=something, args=("happy",))
    t.daemon = True # 메인 기능이 끝나면 같이 끝나게 만드는 기능(자기의 메인)
    t.start()
    print("메인 스레드에서 반복문 시작")
    while True:
        try:
            print("daily...')
            time.sleep(1)
        except KeyBoardInterrupt:
            print("goodbye")
            break

# daily의 메인 쓰레드 안에서 something의 하위스레드가 실행되는것(3초) 메인은 1초

CPU 스케줄링

CPU는 여러개의 프로세스를 빠르게 번갈아 실행하기 위해, 각프로세스를 위해 일하는 시간을 조금씩 나누어 배분한다

• 프로세스 스케줄링: 프로세스의 CPU 점유 순서 및 방법을 결정짓는 일을 가르킴, 알고리즘을 통해 분류

• 프로세스 우선순위

  • 프로세스 별로 요구하는 자원이 상이하기 때문에, 운영체제(스케줄러)가 모든 프로세스에 동일한 빈도로 CPU를 할당한느 것은 비합리적임
  • 운영체제는 프로세스마다 우선순위를 부여하고 이를 기준으로 프로세스를 실행한다.

# ps -el | grep python: mac에서 가능한 현재 돌아가고 있는 프로세스 를 볼수있음(파이썬 이름을 가진)

📝 스케줄링 알고리즘

Queue

운영체제는 준비 상태의 프로세스와 대기 상태의 프로세스를 관리하기 위해 큐(Queue) 자료구조를 사용

• 준비 Q: CPU를 이용하고 싶은 프로세스들이 줄을 서기 위해 존재하는 Q

• 대기 Q: 입출력 장치를 이용하기 위해 대기상태에 있는 프로세스들이 줄을서기 위해 존재

• Q: 프로세스의 정보를 가지고 있는 프로세스 컨트롤 블록들이 줄을서고 운영체제는 줄을 서 있는 컨트롤 블록을 참고해서 어떤 프로세스를 실행할지 결정

스케줄링 알고리즘시 고려사항

• 부하가 최소화되어야 함
• 컴퓨팅 자원을 효율적으로 사용해야 함
• 균형잡힌 스케줄링을 해야 함
• 대기 및 응답 시간이 너무 길어서는 안됨

• 선입선출(First In First Out)
  • 단순하고 직관적이지만 실행시간이 짧은 프로세스가 실행 시간이 긴 프로세스 뒤에서 한참 기다리는 상황이 벌어질 수 있음,
  • (우선순위가 같다면)순서대로 작업하는 것
• 최단 작업 우선(Shortest Job First)
  • 프로세스의 실행 시간을 정확히 예측할수 없다는 문제가 있어 현실적이지 않음
  • (우선순위가 같다면)실행시간이 짧은 것부터 실행
• 라운드 로빈(Round Robin)
  • (우선순위가 같다면)정해진 시간 만큼만 CPU를 점유하고, 시간이 다지나면 컨텍스트 스위칭하는 방식
• 우선순위 스케줄링(Priority Scheduling)
  • 앞선 알고리즘의 기반이 되는 아이디어, 우선순위만 고려할 경우 우선순위가 낮은 프로세스가 배제되어 버리는 '기아' 상태에 빠질 수 있음
  • 각 프로세스마다 우선순위를 부여해서 우선순위를 먼저 처리하는 과정

📝 프로세스 간 통신

• 프로세스는 독립적으로 실행되지만, 필요시 다른 프로세스와 데이터를 주고받으며 통신하는 경우도 있음
• 이를 가르켜 IPC(Inter-Process Coummunication)라 한다

통신을 하려면, 통신의 각주체가 만날 수 있는 매개체가 필요

• 기본적으로 프로세스는 독립적이기에 서로 통신을 할 수가 없음

• 메일슬롯 방식(IPC)(프로세스 간에 관계가 없는 관계에서)
  • 데이터를 받기 위해서 프로세스가 우체통 역할을 하는 객체를 마련하고 이를 통해 데이터를 주고 받는 방식, 단방향이기에 각각 만들어야 함
• 파이프 방식(IPC)(관계가 있는 파일)
  • 익명파이프 또는 네임드 파이프를 통해 데이터를 주고 받는다
  • 익명파이프는 서로 관계가 있는 프로세스 간에 통신을 할떄 사용하는 단방향 파이프
  • 네임드 파이프는 프로세스 간에 양방향 통신을 할떄 사용하는 파이프

# 파이프 기반 통신
from multiprocessing import Process, Pipe
import os

def send(conn):
    print(f"{os.getpid()}가 {os.getppid()}에게 데이터를 보낸다")
    conn.send("Hello Parent!") # 부모 프로세스에게 메시지 전송
    conn.close() # 파이프 닫기

if __name__=="__main__":
    parent, child = Pipe() # 양방향 통신이 가능한 파이프 생성
    p = Process(target=send, args=(child,))
    p.start()
    print("기존 프로세스 아이디:", os.getpid())# 부모 프로세스 ID 출력
    print(parent.recv()) # 자식 프로세스로부터 데이터 받기

    p.join() # 프로세스가 작업을 종료할떄까지 기다림

# 기존프로세스 아이디:
# 자식이 부모에게 보낸다
# hello parent # recv 에서 받은 것

# 실행 흐름
1. 부모 프로세스(main)가 실행되며 Pipe()를 생성
2. 새로운 프로세스(send 함수)가 실행되면서 자식 프로세스가 생성됨
3. 자식 프로세스가 부모에게 데이터 전송 (conn.send())
4. 부모 프로세스가 데이터를 수신 (parent.recv())
5. 자식 프로세스 종료 후 p.join()을 통해 대기 후 종료

--- 추가예제
from multiprocessing import Process
import os 
import time

def writer() :
    print(f"{os.getpid()}가 파일에 쓴다")
    with open("13.txt", "w") as f: # 파이썬의 w는 생성도 포함 없으면 알아서 파일만듬
        f.write("hello")

def reader() :
    print(f"{os.getpid()}가 파일을 읽는다")
    with open("13.txt", "r") as f: 
        print(f.read())

if __name__ == "__main__":
    p1 = Process(target=writer) # 읽을게 있어야 읽으니까 writer먼저
    p1.start()

    time.sleep(2) # 딜레이를 줌으로써 쓰는중에 접근할 수 없게

    p2 = Process(target=reader)
    p2.start()

    p1.join()
    p2.join()

# 7756가 파일에 쓴다
# 7759가 파일을 읽는다
# HELLO

📝 동기화

• 프로세스들은 서로 독립적이지만, 프로세스 간 통신을 하거나 같은 대상에 대한 작업을 함으로써 협력할 수 있다. 그런데 이때, 동시다발적으로 작업을 처리하면 문제가 발생한다.
• 한 파일을 두고 같이 작업을 시행하려 할때 문제 발생
• 프로세스 + 스레드 둘다 적용

• 공유 자원: 프로세스 간 통신에서 공동으로 이용하는 변수가 파일, 입출력 기기등이 존재
  • 각 프로세스의 접근 순서에 따라 결과가 달라질 수 있는데, 프로세스가 동시에 실행할 경우 문제가 발생할 수 있는 영역을 임계구역이라 한다.

상호배제

• 동기화 기법은 임계구역에서 발생할 수 있는 문제를 해결하기 위한 기법
• 동기화 기법 구현 시에는 상호배제라는 조건을 만족시켜야 한다.

• 상호배제란 하나의 프로세스가 임계구역에 들어갔다면 다른 프로세스는 임계구역에 들어갈 수 없다는 조건

뮤텍스 락

• 동시에 접근해서는 안되는 자원에 동시접근 못하도록 막는 동기화 도구
• 임계구역에 접근하는 프로세스는 뮤텍스 락을 통해 다른 프로세스에서 접근하지 못하도록 막을수 있음
• 공유 자원이 하나인 경우
• acquire(): 진입 시 문을 잠그는 함수
• release(): 다쓰고 나올떄 문잠금을 해제하는 함수

from multiprocessing import Process, Value # 프로세스간의 값을 공유하게 만들고 싶을때 사용하는 파이썬 생성자 함수

def counter1(share_num, cnt):
    for i in range(cnt):
        share_num.value += 1

def counter2(share_num, cnt):
    for i in range(cnt):
        snum.value -= 1

if __name__=="__main__":
    shared_number = Value('i', 0)
    p1 = Process(target=counter1, args=(shared_number,5000))
    p1.start()

    p2 = Process(target=counter2, args=(shared_number,5000))
    p2.start()

    p1.join()
    p2.join()

    print("final num is:", shared_num.value)

# 0이 안됨 공유자원이여서 값이 꼬여버림

# 뮤텍스 락을 설정한후 실행
from multiprocessing import Process, Value, Lock # 뮤텍스 락 역할

def counter1(share_num, cnt, lock):
    lock.acquire() # 문잠금
    try :
        for i in range(cnt):
            share_num.value += 1
    finally :
        lock.release()

def counter2(share_num, cnt, lock):
    lock.acquire()
    try:
        for i in range(cnt):
            snum.value -= 1
    finally:
        lock.release()

if __name__=="__main__":

    lock = Lock()1
    shared_number = Value('i', 0)
    p1 = Process(target=counter1, args=(shared_number,5000), lock)
    p1.start()

    p2 = Process(target=counter2, args=(shared_number,5000), lock )
    p2.start()

    p1.join()
    p2.join()

    print("finall num is:", shared_num.value)

# 0으로 잘 나온다.

세마포어(key라 생각)

• 공유자원이 여러개
• 공유자원에 대하여 세마포어 즉 키를 전달해서 공유자원을 못쓰도록 막는것
• wait() : 프로세스가 임계구역에 들어갈 수 있는지 기다려야 하는지 알려주는 용도
• signal() : 임계구역 함수에서 기다리고 있는 함수에서 이제 들어가도 된다고 신호를 보내는 함수

쓰레드 동기화 예시

import threading
from multiprocessing import Value, Lock # 뮤텍스 락 역할

def counter1(share_num, cnt, lock):
    lock.acquire() # 문잠금
    try :
        for i in range(cnt):
            share_num.value += 1
    finally :
        lock.release()

def counter2(share_num, cnt, lock):
    lock.acquire()
    try:
        for i in range(cnt):
            snum.value -= 1
    finally:
        lock.release()

if __name__=="__main__":

    lock = Lock()1
    shared_number = Value('i', 0)
    t1 = threading.Thread(target=counter1, args=(shared_number,5000), lock)
    t1.start()

    t2 = threading.Thread(target=counter2, args=(shared_number,5000), lock )
    t2.start()

    t1.join()
    t2.join()

    print("finall num is:", shared_num.value)

# 0으로 잘 나온다.

📝 데드락(Deadlock)

• 기다리는 대상이 꼬리에 꼬리를 물고 이어져서 아무도 작업을 진행하지 못하는 상황을 데드락이라 한다.
• 공유자원으로 인해 발생

• 데드락 발생조건 4가지 중 모두 만족해야 발생 가능성이 생김
  1. 상호배제(Mutual exclusion) : 공유자원 같이 사용할 수 없어야 한다.
  2. 비선점(No preemption) : 공유자원을 가지고 있을때 빼았아서 사용할 수 없어야 한다
  3. 점유 및 대기(Hold and wait) : 쓰고있을 경우 사용하기 위해 기다리고 있어야 한다.
  4. 원형 대기(Circular waite) : 꼬리에 꼬리를 물어 둥글게 형성되어 있어야 한다.

식사하는 철학자 문제

📌 문제 개요

• N명의 철학자가 원형 테이블에 앉아 있음.
• 각 철학자는 생각(Thinking)하거나 식사(Eating) 함.
• N개의 포크(Fork)가 놓여 있으며, 각 철학자는 왼쪽과 오른쪽의 포크를 사용해야 식사할 수 있음.
• 문제: 모든 철학자가 교착 상태(Deadlock)나 기아 상태(Starvation) 없이 원활하게 식사할 수 있도록 동기화해야 함.

📌 문제 발생 가능성

• 경쟁 조건(Race Condition)

  • 여러 철학자가 동시에 포크를 집으려고 하면 충돌이 발생할 수 있음.

• 교착 상태(Deadlock)

  • 모든 철학자가 왼쪽 포크만 집고 오른쪽 포크를 기다리는 상황이 발생하면 영원히 대기하게 됨.

• 기아 상태(Starvation)

  • 일부 철학자가 계속해서 포크를 획득하지 못하면 식사를 못하고 계속 굶는 상태가 됨.

📌 해결 방법

1. 순서 기반 해결법 (홀수-짝수 철학자)

• 짝수 번호 철학자: 왼쪽 → 오른쪽 포크 순서로 집음
• 홀수 번호 철학자: 오른쪽 → 왼쪽 포크 순서로 집음
  📌 장점: 교착 상태(Deadlock) 방지 가능

2. 웨이터(Butler) 알고리즘

• 중앙 관리자(웨이터, Mutex)가 철학자들에게 최대 (N-1)개까지만 포크를 허용
• 철학자가 식사하려면 웨이터에게 허가 요청을 해야 함
  📌 장점: Deadlock 예방 및 자원 사용 관리 가능

3. 개별 철학자 스레드 + 세마포어

• 각 철학자 스레드가 세마포어(Semaphore)를 사용하여 포크를 집도록 제한
• 두 개의 세마포어를 확보한 철학자만 식사 가능

📌 파이썬 코드 예제 (Semaphore 사용)

import threading
import time

N = 5  # 철학자 수
forks = [threading.Semaphore(1) for _ in range(N)]  # 포크 세마포어 배열

def philosopher(i):
    left = i
    right = (i + 1) % N

    while True:
        print(f"철학자 {i} 생각 중...")
        time.sleep(1)

        # 포크 집기
        forks[left].acquire()  # 왼쪽 포크 획득
        forks[right].acquire()  # 오른쪽 포크 획득

        print(f"🍽️ 철학자 {i} 식사 중...")
        time.sleep(2)

        # 포크 내려놓기
        forks[left].release()
        forks[right].release()

        print(f"철학자 {i} 식사 완료. 다시 생각 시작...")

# 철학자 스레드 생성 및 실행
threads = []
for i in range(N):
    t = threading.Thread(target=philosopher, args=(i,))
    threads.append(t)
    t.start()

# 모든 스레드가 종료될 때까지 대기
for t in threads:
    t.join()

📌 코드 설명

1. threading.Semaphore(1): 각 포크를 이진 세마포어로 설정 (한 번에 한 철학자만 사용 가능)
2. 각 철학자는 왼쪽과 오른쪽 포크를 집고 식사
3. 식사가 끝나면 포크를 놓고 다시 생각 상태로 전환
4. 스레드 기반 실행을 통해 병렬적으로 실행

📌 해결된 문제점

• Deadlock 방지

  • 두 개의 포크를 동시에 획득한 경우만 식사가 가능하여 교착 상태 예방

• Starvation 해결 가능

  • 웨이터 알고리즘 등 추가 적용 가능

• 경쟁 조건 방지

  • Semaphore.acquire()를 사용하여 한 번에 하나의 철학자만 포크를 획득

📌 결론

• 식사하는 철학자 문제는 동기화(Synchronization)의 대표적인 문제
• 세마포어, 웨이터 알고리즘 등을 활용하여 Deadlock과 Starvation을 방지 가능
• 멀티스레딩 환경에서 공유 자원(포크) 관리 방법을 학습하는 데 중요한 개념

📝 메모리

• 레지스터
  • 가장 빠른 대신 용량이 적음
  • 휘발성
• 캐시(L1/L2)
  • CPU와 RAM사이에서 중간 저장소
  • CPU에 가장 근접해 있는 메모리
  • 레지스터와 RAM사이에 데이터 이동이 있을때 속도를 보완하는 역할을 하는 메모리
  • CPU 입장에선 바로바로 데이터를 꺼내어 쓸쑤 있는 저장소 같은 느낌
  • 휘발성
• 메인 메모리(RAM)
  • 운영체제와 프로세서 들이 올라가는 공간
  • 실행중인 프로그램을 올리기 위해 주로 사용
  • 휘발성
• 보조기억장치(HDD/SSD)
  • 크고작은 파일들을 저장하기 위한 용도
  • 속도는 느리지마 용량은 큼
  • 비휘발성

메모리 관리 방식

가변 분할 방식

• 프로세스의 크기에 따라 가변적으로 메모리 할당
• 연속메모리 할당
• segmentation

• 스와핑
  • 실행 상태의 프로세스는 메인 메모리에 올리고, 실행 상태가 아닌 프로세스들은 보조 장치에 따로 마련된 스왑 영역에 올린 다음 프로세스의 상태 변화에 따라 두공간 사이에서 프로세스가 이동하는것을 가리켜 스와핑 이라한다.

• 외부 단편화
  • 프로세스를 메모리에 연속 배치하는 작업을 하다보면 메모리가 낭비되는 '외부 단편화' 상황이 발생하기도 함
  • 기존의 메모리 들의 작업이 끝나고 분할되어 있는 공간은 충분하지만 쪼개어져 있어 외부의 큰 메모리를 요하는 작업을 실행할 수가 없을때 외부 단편화라 한다.
  • 조각 모음 : 현재 실행중인 작업을 중단하고 빈공간을 모아서 다시 실행 하는 것을 의미함, 시스템에 큰 부하를 주어 빈번하게 사용하면 안됨

Garbage Collection 예제

# 문자열 객체를 변수 my_name이 참조했다
# 레퍼런스 카운트가 1인 상태
my_name = "Gookhee"

# 레퍼런스 카운트가 2인 상태 : 국희라는 이름을 참조할수 있는 변수가 2개란 뜻
your_name = my_name

my_name = 1 # 레퍼런스 카운트 1
your_name = 2 # 레퍼런스 카운트 0

# 레퍼런스 카운트가 0이된 것을 제거 대상이라 생각한다
# 이 것을 Garbage Collection 이라 생각한다 : 소멸 대상으로 등록

let players ={
    boys :{
        Bregkamp: "Striker"
    }
}

let persons = players

players = ["Son", "Park"]

let human = persons.boys

persons = "persons"

human = null // 비로소 레퍼런스 카운트 0 -> 소멸대상 -> 시스템의 여유가 생길때 삭제

고정 분할 방식

• 메모리를 고정된 크기로 나누어 놓고 프로세스마다 필요한 칸만큼 할당
• 비연속 메모리 할당
• paging

• 외부 단편화

  프로세스를 메모리에 연속 배치하는 작업을 하다보면 메모리가 낭비되는 '외부 단편화' 상황이 발생하기도 함

• 프로세스가 공간을 차지고 있다 작업이 종료되지만 공간이 쪼개

📝 가상 메모리 관리

작은 메인 메모리(RAM)이 프로세스를 실행한다?, 그렇다면 실행할 프로세스가 메인 메모리 보다 덩치가 큰경우엔 스와핑을 활용해 사용

페이징

• 고정 분할 방식
• 페이징 기법에서는 메모리 공간을 일정한 크기의 페이지로 나누어 다룬다.
• 페이지 안에서 빈공간이 남는 내부 단편화가 발생하는 문제

• 논리 주소 공간

  • 사용자와 프로세스가 참조하는 공간
  • 실제 메모리보다 더 큰 공간
  • 보조 기억장치의 크기를 지원받아서 크기를 키운것

• 물리 주소 공간

  • 실제 메모리의 공간

• CPU가 논리주소를 통해 메모리에 접근하고 이를 물리주소로 변환하여 실제메모리에 올리는 것으로 프로세스를 실행 함

• 페이지 테이블

  • 페이지 번호와 프레임 번호를 짝지어 주는 테이블
  • CPU내에는 각프로세스 테이블이 적재된 주소를 가리키는 페이지 테이블 베이스 레지스터가 있음

• 페이지 테이블 엔트리

  • 접근비트: 페이지가 메모리에 올라온 후 데이터에 접근이 있었는지, 1은 접근이 있었다 반대는 0
  • 변경비트: 페이지가 메모리에 올라온 후 데이터의 변경이 있었는지, 1은 접근이 있었다 반대는 0
  • 유효비트: 페이지가 어디에 있는지, 0이라면 스왑영역, 1이라면 물리영역에 있다. -> 메모리에 올라가 있지 않은 영역 즉 스왑영역에 접근하려 한다면 페이지폴트 라는 예외가 발생한다 -> 다른 처리가 필요
  • 보호비트: 페이지에 대한 읽기, 쓰기, 실행 권한이 어떻게 되는지

• 페이지드 세그멘테이션

  • 가변 분할 방식과 고정 분할 방식을 혼합한 가상 메모리 관리 방식
  • 두 방식의 단점을 보완 - 세그먼트를 다시 페이지로 나누어 관리

foods =["햄버거", "샐러드", "비스킷"]

print(id(foods)) # 논리 메모리 주소 확인

mv = memoryview(b"happy day") # b: 바이트 형식으로 저장하겠다.

print(mv)

print(mv[0])
print(mv[1])
print(mv[2]) # 같은 p니까 같은 값 도출
print(mv[3])

# 메모리 사용 현황

import psutil
import os

print("메모리 사용량 조회하기")

memory_dict = dict(psutil.virtual_memory()._asdict()) # 시스템 사용량에 대해 튜플형식으로 반환해줌
print(memory_dict)

total = memory_dict["total"] # 메모리 총량
availabe = memory_dict["available"] # 메모리 즉시 제공 가능량
percent = memory_dict["percent"] # 메모리 사용률

pid = os.getpid()
current_process = psutil.Process(pid)

kb = current_process.memory_info()[0] / 2 ** 20 # 메모리 사용량

---

import tracemalloc # 메모리가 어떻게 할당되는지 추적하는 것, python 3.4 ~~

tracemalloc.start()

# 메모리 할당이 진행되는 작업 아무거나 
data = [b'%d' % num for num in range(1, 10001)]
joined_data = b' '.join(data)

current, peak = tracemalloc.get_traced_memory()
print(f"현재 메모리 사용량 : {current / 10 ** 6} MB")
print(f"최대 메모리 사용량 : {peak / 10 ** 6} MB")

tracemalloc.stop()

tracemalloc.get_tracemalloc_memory() # 트레이스 말록 사용하느라 사용한 메모리 확인
print(traced / 10 ** 6)

📝 페이지 교체

요구 페이징

• CPU가 특정 페이지에 접근하는 명령어를 실행 햇을때, 해당 페이지가 스왑 영역에 있어서 당장 실행시킬 수 없는 상태일 경우에는 '페이지 폴트' 예외가 발생한다.
• 페이지 폴트 예외가 발생하면 스와핑 작업이 먼저 진행된 후에 프로세스가 실행된다.
• 실행할 모든 프로세스를 메모리에 올려두는 것은 시스템에 부담이 될 수 있는 만큼 당장 필요한 페이지만을 메모리에 우선 적재하는 방법을 가리켜 '요구페이징' 이라 한다

• 페이지 폴트시 -> 페이지 교체 정책 대표적인 3가지
  • 선입선출: 가장 오래된 페이지 교체
  • 최적 페이지 교체: 자주 사용될 예정인 페이지를 내버려두고 사용이 거의 안될 예정인 페이지를 교체, 실제로 구현하기 어려움 예상이 어려워서
  • LRU(Least Recently Used): 최근 사용빈도가 가장 적은것을 교체하는 것, 구현하기 쉽고 성능도 좋음

# 선입선출 하나 만들어보기 운영체제라 생각하고
class PageReplacementFIFO:
    def __init__(self, capacity):
        self.capaciry = capacity
        self.pages = []

    def access_page(self, page) : # 4개째부터 교체가 이루어지게 하자
        if page not in self.pages : # 기존의 페이지에 없으면 집어넣자
            if len(self.pages) >= self.capacity:
                self.pages.pop(0) # 제일 앞에 있는 페이지 제거
            self.pages.append(page)
    def status(self):
        print("현재 페이지 상태 :", self.pages)

page_replacement = PageReplacementFIFO(capacity = 3)
page_replacement.status() # 빈상태 []
page_replacement.access_page(3)
page_replacement.status() # [3]
page_replacement.access_page(2)
page_replacement.status() # [3,2]
page_replacement.access_page(1)
page_replacement.status() # [3,2,1]

page_replacement.access_page(4)
page_replacement.status() # [2,1,4]

스래싱(thrashing)

• 모든것의 근본적인 이유는 메모리 공간의 부족, 즉 프레임 부족 때문
• 프레임이 부족하면 페이지 볼트가 자주 발생
• 페이지 폴트가 발생하면 잦은 스와핑 작업으로 인해 CPU 사용률이 떨어지게 된다.
• CPU 사용률이 떨어지면 운영체제는 더 많은 프로세스를 메모리에 올리려 하고, 이는 더 잦은 페이지 폴트로 이어져 악순환에 빠지게 된다.
• 이러한 문제를 스레싱이라고 한다.

📝 스래싱(Thrashing) 개요

• 스래싱(Thrashing) 은 가상 메모리 시스템에서 페이지 부재(Page Fault)가 너무 자주 발생하여 CPU가 메모리 스왑 처리에만 집중하고 실제 작업을 거의 수행하지 못하는 상태를 의미함.
• 스래싱이 발생하면 성능이 급격히 저하되며, 시스템이 비효율적으로 동작함.

📌 스래싱 해결책

📌 워킹셋(Working Set) 기법

1. 개념

• 워킹셋(Working Set) 이란 프로세스가 일정 시간 동안 자주 참조하는 페이지들의 집합을 의미함.
• 가상 메모리에서 프로세스가 일정 기간 내 참조한 페이지를 추적하여 자주 사용하는 페이지를 메모리에 유지하고, 사용하지 않는 페이지를 제거함으로써 스래싱을 방지.

2. 워킹셋의 정의

• W(Δ): 특정 시간 구간 Δ(델타) 동안 프로세스가 참조한 페이지 집합
• Δ 값 선택이 중요
  • Δ 값이 너무 작으면 → 너무 많은 페이지가 제거됨 (과도한 페이지 부재)
  • Δ 값이 너무 크면 → 불필요한 페이지까지 유지됨 (메모리 낭비)

3. 워킹셋을 활용한 스래싱 방지

• 워킹셋을 유지하며 최소한의 프레임을 확보하여 스래싱을 방지.
• 일정 시간 동안 참조되지 않은 페이지는 스왑 아웃(Swap Out)하여 메모리 활용도를 높임.

4. 워킹셋 기법의 장점

✔ 스래싱 방지: 자주 사용하는 페이지만 유지하여 CPU가 불필요한 스왑을 하지 않도록 함
✔ 메모리 효율 증가: 실제 필요한 페이지만 유지하므로 불필요한 페이지 사용 감소
✔ 동적 프레임 조정 가능: 프로세스의 실행 패턴에 따라 필요한 페이지를 동적으로 조절 가능

📝 파일

의미있는 정보들을 한덩어리로 모아 둔 논리적 단위, 보저기억 장치에 저장, 실행시 메인메모리 사용

• 파일이름, 파일 실행에 필요한 정보, 메타데이터

• 확장자 : 운영체제에게 있어 중요한 정보
• 응용 프로그램이 파일에 접근하려면 시스템 호출을 제공받아야 한다.

with open("number_one.txt", "w") as f:
    f.write("one")
with open("number_two.txt", "w") as f:
    f.write("two")
with open("number_three.txt", "w") as f:
    f.write("three")
with open("number_four.txt", "w") as f:
    f.write("four")

import glob # 파일네임의 패턴을 이용해 한꺼번에 접근

for filename in glob.glob("*.txt", recursive= True): 
    print(filename)

import fileinput # 내용들도 한번에 처리하기

with fileinput.input(glob.glob("*.txt")) as fi:
    for line in fi :
        print(line) # 모든 파일 내용 읽기

import fnmatch # filenamematch
import os

for filename in os.listdir('.'): # listdir: 현재 디렉토리에 있는것을 찾음
    if fnmatch.fnmatch(filename, "??????_one.txt") :
        print(filename) # number_one.txt

---

# 파일 관련 예외는 운영체제와 관계가 있음
try : 
    f = open("none.txt", "r") # 존재하지 않는것에 r을 하면 예러
    print(f.read())
    f.close()
except FileNotFoundError as e:
    print(e)
    issubclass(FileNotFoundError, OSError) # 첫번쨰 인자가 두번쨰 인자의 하위인자가 맞는지

const fs = require("fs")
fs.readFIle("number_one.txt", "utf8", (err, data) => {
    if(err){
        console.log("파일을 읽는 도중 오류가 발생.", err)
        return;
    } 
    console.log("파일 내용:", data)
})

let content = "four four four"
fs.writeFile("number_four.txt", content, (err) => {
    if(err) {
        console.log("파일을 쓰는 도중 오류가 발생.", err)
        return;
    }
    console.log("파일쓰기 완료")

})

let content = "four four four"
fs.appendFile("number_four.txt", content, (err) => {
    if(err) {
        console.log("파일을 쓰는 도중 오류가 발생.", err)
        return;
    }
    console.log("파일쓰기 완료")

})

📝 디렉터리

• 운영체제는 파일을 정돈할 수 있도록 디렉터리를 지원
• 디렉터리도 파일의 일종
• 디렉터리에 저장된 파일 정보는 테이블 형태로 관리
• 테이블 행 하나하나를 가리켜 디렉터리 엔트리 라고 부른다

# os 파일 시스템 관련 함수
# pwd를 통해 현재 파일위치를 찾는다
pwd = "경로 복붙"

filenames = os.listdir(pwd) # 경로상의 파일들 반환 

# 디렉터리인지 아닌지 여부
print(os.path.isdir(filenams[0])) # False
print(os.path.isdir(pwd + "/폴더") # True

# 파일인지 아닌지
print(os.path.isfile(filenams[0])) # True
print(os.path.isfile(pwd + "/폴더") # False

# 파일이름과 확장자 분리
filepath = pwd + "/" + filenames[0]
print(filepath)
os.path.splitext(filepath)
name, ext = os.path.splitext(filepath)
print(ext) # .txt

---
# 경로와 확장자 이요해 파일 찾고, 내용 읽기
import os
def searchFile(dirname, extension):
    filenames = os.listdir(dirname)
    for filename in filenames:
        filepath = os.path.join(dirname, filename)
        if os.path.isdir(filepath):
            searchFile(filepath, extension)
        elif os.path.isfile(filepath) :
            name, ext = os.path.splitext(filepath) :
            if ext == extension :
                with open(filepathm, "r", encoding="utf-8") as f:
                    print(f.read())

searchFile("pwd경로", ".js")  # 자바스크립트 코드 읽음

---
# 파일 복사 또는 이동
import os
import shutil

pwd = "/"
filenames = os.listdir(pwd)

for filename in filenames:
    if "tokyo" in filename :
        origin = os.path.join(pwd, filename)
        print(origin)

        shutil.copy(origin, os.path.join(pwd, "copy.txt")) # 원본 , 어디다가, 
        shutil.move(origin, os.path.join(pwd, "anony")) # 파일 이동

---
# 파일 경로를 문자열이 아닌 객체로 다루기
import os
import pathlib

for p in pathlib.Path.cwd().glob("*.txt"): # 현재 워킹디렉토리 cwd current working directory
    print(p.parent) # 해당 텍스트들의 부모 경로가 나옴
    new_p = os.path.join(p.parent, p.name)
    print(new_p) # 내 파일의 총 경로

📝 파일과 메모리

• 하나의 파일은 여러개의 블록으로 이루어져 있다.
• 블록을 메미로에 할당할 때는 연속방식 또는 불연속방식을 사용할 수 있다.

🔗 연결 리스트 (Linked List) 자료구조

📌 개요

연결 리스트(Linked List)는 메모리의 불연속 할당을 기반으로 데이터를 저장하는 선형 자료구조입니다.
연결 리스트는 데이터를 노드(Node) 단위로 관리하며, 각 노드는 데이터와 다음 노드를 가리키는 포인터(Pointer)를 가지고 있습니다.

📌 연결 리스트의 구조

• 연결 리스트는 노드(Node)라는 단위로 구성됨
• 각 노드는 데이터 영역과 다음 노드를 가리키는 포인터로 구성됨
• 메모리에서 연속적이지 않은 영역을 사용 (불연속 메모리 할당)

🔹 기본적인 연결 리스트의 형태:

Head
 |
 ▼
[Data | Next] -> [Data | Next] -> [Data | Next] -> NULL

📌 연결 리스트의 특징

📌 연결 리스트의 종류

1. 단일 연결 리스트 (Singly Linked List)

• 한 방향으로만 다음 노드를 가리키는 포인터를 가짐
• Head 노드부터 순차적으로 접근 가능

2. 이중 연결 리스트 (Doubly Linked List)

• 양방향(앞, 뒤)으로 접근 가능한 포인터를 가짐
• 노드 탐색과 삭제가 더 쉬움

3. 원형 연결 리스트 (Circular Linked List)

• 마지막 노드가 첫 번째 노드를 가리키도록 구성
• 순환적인 데이터 구조를 나타낼 때 사용

📌 연결 리스트의 장점과 단점

장점

• 삽입과 삭제가 빠름 (O(1) 가능)
• 메모리 동적 할당으로 유연한 메모리 관리 가능
• 메모리의 낭비가 적음 (필요한 만큼만 사용)

단점

• 임의 접근(Random Access)이 불가능 (순차 접근만 가능)
• 탐색(Search)에 시간이 오래 걸림 (O(n))

📌 연결 리스트와 배열의 비교

📌 결론

• 연결 리스트는 불연속 할당 방식을 사용하는 선형 자료구조로, 포인터를 이용하여 데이터 간 연결을 유지합니다.
• 삽입과 삭제가 효율적이며 동적 메모리 관리가 중요한 경우에 특히 유용합니다.
• 탐색 속도는 배열보다 느리기 때문에 상황에 따라 적합하게 선택하여 사용해야 합니다.

📌 전체 코드 (C언어)

#include <stdio.h>
#include <stdlib.h>

// 노드 구조체 정의
typedef struct Node {
    int data;
    struct Node* next;
} Node;

// 연결 리스트 함수들
void insertFront(Node** head, int value);
void deleteNode(Node** head, int key);
void printList(Node* head);

// 메인 함수
int main() {
    Node* head = NULL;  // 초기 연결 리스트는 비어 있음

    // 노드 추가
    insertFront(&head, 10);
    insertFront(&head, 20);
    insertFront(&head, 30);

    printf("연결 리스트 출력: ");
    printList(head);

    // 노드 삭제
    deleteNode(&head, 20);
    printf("노드 삭제 후 출력: ");
    printList(head);

    return 0;
}

// 연결 리스트의 맨 앞에 노드 삽입
void insertFront(Node** head, int value) {
    Node* newNode = (Node*)malloc(sizeof(Node));
    newNode->data = value;
    newNode->next = *head;
    *head = newNode;
}

// 특정 값 가진 노드 삭제
void deleteNode(Node** head, int key) {
    Node* temp = *head, *prev = NULL;

    if (temp != NULL && temp->data == key) {
        *head = temp->next;
        free(temp);
        return;
    }

    while (temp != NULL && temp->data != key) {
        prev = temp;
        temp = temp->next;
    }

    if (temp == NULL) return;

    prev->next = temp->next;
    free(temp);
}

// 연결 리스트 출력
void printList(Node* head) {
    Node* temp = head;
    while (temp != NULL) {
        printf("%d -> ", temp->data);
        temp = temp->next;
    }
    printf("NULL\n");
}

📌 실행 결과 예시

연결 리스트 출력: 30 -> 20 -> 10 -> NULL
노드 삭제 후 출력: 30 -> 10 -> NULL

• 30 -> 20 -> 10 -> NULL 순서로 추가됨
• 20을 삭제한 후 30 -> 10 -> NULL이 출력됨

📝 데이터 표현

• 컴퓨터는 모든 정보를 0과 1로 표현한다.
• 컴퓨터의 저장장치가 비트 단위로 구성되어 있기 때문이다
  • 비트란, On과 Off상태만 표현할 수 있는 정보 단위이다.

• 부동 소수점: 컴퓨터에서 실수(Real Number)를 표현하는 방식

  • 부호(Sign): 양수(0) 또는 음수(1)를 표현하는 부분
  • 지수부(Exponent): 소수점 위치를 나타내는 부분
  • 가수부(Mantissa or Fraction): 숫자의 정밀도를 나타내는 부분

• 컴퓨터가 읽고, 표현할 수 있는 문자의 모음인 문자 셋 charcter set 이 있다

  • 문자 인코딩
  • ex) UTF-8, 아스키코드, ...

📝 스레드 풀링

• 스레드는 프로세스의 작업을 나누어 관리하는 유용한 도구지만 스레드가 너무 많아지는 것은 바람직하지 않다.
• 스레드의 생성과 소멸은 시스템에 많은 부담을 준다.
• 스레드 풀은, 특정 개수의 스레드가 여러 개의 일을 차례대로 수행하는 기법이다.

# 개념은 같은 프로세스 풀 예제
import concurrent.futures
import time

def task(name) :
    time.sleep(0.5)
    return f"{name}의 작업 완료"

if __name__ == "__main__" :

    with concurrent.futures.ProcessPoolExecutor(max_workers=3) as executor :
        future_name = {executor.submit(task, f"Task-{i}") : f"Task-{i}" for i in range(5)}
        print(future_name)
        # 작업 완료된 순서대로 결과 출력, 작업이 들어간 순서대로 처리하지 않을 수도 있음
        for future in concurrent.futures.as_complted(future_name):
            name = future_name[future]
            try :
                result = future.result()
                print(f"{name}의 결과 : {result}")
            except Exception as e:
                print(e)

📝 라운드로빈 모델

선입 선출이지만 정해진 시간 만큼만 일하는 것

def roundrobin(processes, busrt_time, time_quantum) :
    n = len(processes)
    remaining_time = list(burst_time)
    turnaround_time = [0] * n
    waiting_time = [0] * n

    time = 0
    queue = []

    while True :
        all_completed = True # 모든 프로세스 종료 시 반복문 종료를 위한 플래그

        for i in range(n) :
           if remaining_time[i] > 0:
               all_completed = False

               if remaining_time[i] > time_quantum :
                   time += time_quantum
                   remaining_time[i] -= time_quantum
                   queue.append(i)
                else :
                    time += remaining_time[i]
                    turnaround_time[i] = time
                    remaining_time[i] = 0
                    waiting_time[i] = turnaround_time[i] - burst_time[i]
        if all_completed:
            break

    print("Process\tTurnaround Time\tWaiting Time")
    for i in range(n):
        print(f"P{i+1}\t\t{turnaround_time[i]}\t\t{waiting_time[i]}")

# 함수 호출
processes = [1, 2 ,3]
burst_time = [10, 5, 8]
time_slice = 2

roundrobin(processes, burst_time, time_slice)

배포 Deployment

• 배포란?

  • 개발된 코드를 사용자들이 접근할 수 있도록 서버에 배치하는 과정
  • 로컬 환경에서 실행되던 코드를 프로덕션(운영) 환경으로 이동
  • 배포 후에도 유지보수와 모니터링 필요

• 배포 환경의 유형

  • 로컬 환경 (Local Environment): 개발자 로컬 컴퓨터에서 실행
  • 스테이징 환경 (Staging Environment): 실제 배포 전에 테스트하는 환경
  • 프로덕션 환경 (Production Environment): 최종 사용자들이 사용하는 실제 환경

배포의 주요 단계

1. 코드 준비: 로컬에서 실행 가능한 상태로 코드 작성 및 검증
2. 빌드/패키징: 코드와 의존성을 묶어 실행 가능한 형태로 준비
3. 환경 구성: 서버에서 실행에 필요한 설정 파일 작성 (환경 변수, 포트 설정)
   • 환경 파일을 만들어 감춰나야 예를들어 access토큰 같은 탈취를 안당함
4. 배포 및 검증: 프로덕션 서버에서 실행 및 테스트

배포 방식 비교: 전통적인 방식 vs 최신 방식

• 전통적인 배포 방식: 서버 직접 설정 (On-Premise)

  • 특징
    • 직접 설정 가능: 개발자가 환경을 자유롭게 구성 가능
    • 추가 비용 없음: 직접 서버를 관리하여 추가 서비스 비용이 적음
  • 단점
    • 환경 불일치 문제: 개발 환경과 운영 환경이 다르면 오류 발생
    • 수동 배포 비효율성: 변경 사항이 있을 때마다 직접 서버에 접속해서 배포해야 함
    • 확장성 부족: 트래픽이 많아질 경우 서버 성능을 조정하는 것이 어려움

• 최신 배포 방식: Docker, CI/CD, 클라우드 플랫폼을 활용한 자동화 배포 Cloud-based, Serverless

  • 특징
    • 컨테이너(ex. Docker) 사용: 개발 환경과 운영 환경을 동일하게 유지
    • CI/CD 자동화
    • 클라우드 서비스 (AWS, GCP, Azure, NCP 등) 이용: 확장성 보장
  • 배포 자동화 예시
    • Github -> Cloudtype으로 자동 배포
    • 코드 푸시(git push)만 하면 자동으로 배포

클라우드 서비스 장단점

클라우드 서비스 선택 시 고려사항

• 프로젝트 규모 및 복잡도

  • 소규모 프로젝트: Render, Railway
  • 대규모 확장 가능한 프로젝트: AWS, GCP, Azure

• 예산

  • 무료 티어 제공 여부 확인: Fly.io, Render, Railway 등
  • 높은 트래픽이나 데이터베이스 사용이 예상되면 비용 분석 필수

• 설정의 간편성

  • 빠른 설정: Heroku, Render 추천
  • 복잡한 설정과 유연성을 원한다면 AWS, GCP 추천

• 기술 스택 호환성

  • 정적 사이트 중심: Vercel, Netlify
  • 서버리스 애플리케이션: AWS Lambda, GCP CloudFunction
  • 데이터베이스 중심: Railway, AWS RDS

• 서비스 안정성 및 지원

  • 글로벌 서비스, 데이터 센터: AWS, GCP, Azure 추천
  • 지역 기반 서비스: 데이터 센터 위치를 확인할 것

• 성능 및 확장성

  • 예측 가능한 트래픽: Render, Vercel
  • 높은 확장성 요구: AWS, GCP, Azure

• CI / CD 및 개발자 도구

  • 자동화 배포: Render, Vercel, Netilfy 추천
  • 데이터 분석, AI 통합이 필요하다면 GCP 추천

DevOps와 배포 자동화

• DevOps: Development + Operations

  • 소프트웨어 개발과 운영을 하나로 통합하여 빠르고 안정적인 배포를 가능하게 하는 문화 및 기술
  • 자동화 -> 안정성 증가 및 빠른 업데이트 가능
  • CI(Continuous Integration) 지속적 통합: 코드 변경 사항을 자동으로 빌드 및 테스트
  • CD(Continuous Deployment) 지속적 배포: 코드 변경 사항을 운영 환경에서 자동 배포

• CI/CD 파이프라인 예시

  • 개발자가 코드 변경 후 github에 푸시
  • CI 시스템 (Github Actions)이 자동으로 빌드 및 테스트 실행
  • 테스트 통과 시, CD 시스템 (Cloudtype)이 배포 자동화 진행
  • 배포 완료 후, 운영 서버에서 서비스 활성화

배포의 확장성과 유지보수

• 배포 시스템이 확장성을 가지려면?

  • MSA(Micro Service Architecture): 여러 작은 서비스로 분리
  • Container Orchestration (Kubernetes) 사용: 자동 확장 및 관리 지원
  • 무중단 배포 (Blue-Green Deployment): 운영 중에도 장애 없이 배포 (프론트면 실제 파일에 덮어 씌워 버리는 것)

• 배포 후 지속적인 모니터링

  • 서비스 상태 체크: Health Check API
  • 로깅 및 오류 분석: Sentry, Prometheus
  • 성능 모니터링: Grafana

Docker란

• 애플리케이션을 컨테이너라는 가상 환경에서 실행하는 기술
  • 컨테이너를 사용하면 OS에 영향을 받지 않고 어디서든 동일한 환경에서 실행 가능
    • A는 Windows에서, B는 CentOS에서 프로그래밍을 할 때 불일치가 발생할 수 있으며, 이 때 컨테이너 기술을 사용하면 둘이 동일한 환경에서 서버 구동 가능

컨테이너 vs 가상머신 (VM)

가상머신이란 ?

하드웨어 자원을 가상화하여 독립적인 운영체제를 실행할 수 있도록 만든 소프트웨어 환경을 의미해. 즉, 하나의 물리적인 컴퓨터 위에서 여러 개의 가상 컴퓨터를 실행

• 특징

  1. 독립적인 운영체제 실행
     • VM마다 각각의 운영체제(Windows, Linux 등)를 설치할 수 있음.
     • 호스트 OS(기본 운영체제)와 다른 OS도 설치 가능.
  2. 하드웨어 리소스 분할
     • CPU, RAM, 저장소 등의 자원을 가상화하여 할당.
     • 하나의 물리 서버에서 여러 VM을 실행할 수 있음.
  3. 보안 및 격리성
     • 각 VM은 서로 완전히 독립적이라, 하나의 VM이 문제가 생겨도 다른 VM에 영향을 주지 않음.
  4. 유연한 배포와 관리
     • 클라우드 서비스(AWS, Azure, GCP)에서 가상머신을 쉽게 생성 및 관리 가능.

• VM의 대표적인 종류

  • VirtualBox → 로컬 환경에서 VM 실행 (무료)
  • VMware → 기업에서 많이 사용 (유료)
  • Hyper-V → Windows 전용 가상화 소프트웨어
  • AWS EC2, GCP Compute Engine → 클라우드 기반 가상머신

• 가상머신(VM)은 언제 사용하면 좋을까?

  • 서로 다른 OS 환경을 실행해야 할 때 (예: Windows에서 Linux 실행)
  • 보안이 중요한 애플리케이션을 실행할 때 (완전한 격리가 필요할 때)
  • 물리적 서버를 여러 개로 나눠서 사용하고 싶을 때

도커

• 핵심

  • 이미지(image): 실행 가능한 애플리케이션 패키지
  • 컨테이너(Container): 실행 중인 이미지 (실제 애플리케이션이 실행됨)
  • Dockerfile: Docker 이미지를 생성하는 설정 파일

• 주요 명령어

  • 실행 중인 컨테이너 목록 조회: docker ps
  • 모든 컨테이너 목록 (중지된 컨테이너 포함): docker ps -a
  • 실행 중인 컨테이너 중지: docker stop <container_id>
  • 실행 중인 컨테이너 강제종료: docker kill <container_id>
  • 컨테이너 삭제: docker rm <container_id>
  • 컨테이너 로그 확인: docker logs <container_id>

이미지 빌드 & 실행

• 도커 이미지 빌드
  • docker build -t <이미지명><태그>, (docker build -t app .)
• 도커 컨테이너 실행 (포트 매핑 포함)
  • docker run -d -p <호스트포트>:<컨테이너포트> <이미지명>

이미지 및 볼륨 관리

• 모든 도커 이미지 목록 확인
  • docker images
• 특정 도커 이미지 삭제
  • docker rmi <image_id>
• 사용하지 않는 이미지 및 볼륨 정리
  • docker system prune -a

Docker 컨테이너 최적화

• 실행 환경을 최대한 가볍게 만들어야 함

  • 작은 베이스 이미지 사용
    • FROM python:3.10 (922MB)
    • FROM python:3.10-alpine (25MB)

• 불필요한 캐시 저장 방지 & 빌드 속도 최적화

  # 1. Python 3.10 기반 이미지 사용
  FROM python:3.10-alpine (alpine사용 경량화)
  

2. 작업 디렉토리 생성

WORKDIR /app

3. 의존성 설치

COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt

4. FastAPI 애플리케이션 복사

COPY . .

5. 컨테이너 실행 명령어 설정

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

* 다중 스테이지 빌드 적용
```python
# 1. 빌드 스테이지 (의존성 설치만 수행)
FROM python:3.10-alpine (alpine사용 경량화)
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 2. 실행 스테이지 (필요한 파일만 복사)
FROM python:3.10-alpine (alpine사용 경량화)
WORKDIR /app
COPY --from==builder /app /app
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

• .dockerignore 활용

  • pycache/
  • .env
  • *.log

• FastAPI Docker 이미지 빌드 예시

  # main.py
  from fastapi import FastAPI
  

app = FastAPI()

@app.get("/") async def read_root(): return {"message": "FastAPI on Docker!"}

requirements.txt

fastapi uvicorn

Dockerfile

1. Python 3.10 기반 이미지 사용

FROM python:3.10-alpine

2. 작업 디렉토리 생성

WORKDIR /app

3. 의존성 설치

COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt

4. FastAPI 애플리케이션 복사

COPY . .

5. 컨테이너 실행 명령어 설정

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

### Docker Compose로 FastAPI + DB 연동

* Docker Compose: 여러 개의 Docker 컨테이너를 한 번에 실행하는
도구
* 주로 서버 + DB 등을 한 번에 돌릴 때 사용
  - ex) FastAPI + PostgreSQL + Redis
* `docker-compose up -d`
```yaml
# compose.yaml
version: "3.9" # Docker Compose 버전

services:
  db:
    image: postgres: 13 # PostgreSQL 13 버전 사용
    container_name: fastapi_postgres
    restart: always # 컨테이너가 종료되면 자동 재시작
    environment:
      POSTGRES_USER: fastapi_user
      POSTGRES_PASSWORD: fastapi_password
      POSTGRES_DB: fastapi_db
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data # DB 데이터 저장

  fastapi:
    build: . # 현재 디렉토리의 Dockerfile를 사용해 이미지 빌드
    container_name: fastapi_app
    depends_on:
      - db # db 서비스가 실행된 후 fastapi 서비스 실행
    ports:
      - "8000:8000"
    environment:
      DATABASE_URL: "postgresql://fastapi_user:fastapi_password@db:5432/fastapi_db"

volumes:
  postgres_data: # PostgreSQL 데이터를 저장할 볼륨(컨테이너 재시작 시 데이터 유지)

도커 실습

# (1) 도커 이미지 빌드 (현재 디렉토리의 Dockerfile 사용)
docker build -t app .

# (2) 빌드된 이미지 확인
docker images

# (3) 컨테이너 실행 (-d: 백그라운드 실행, -p: 포트 매핑)
docker run -d -p 8000:8000 app


# (1) 실행 중인 컨테이너 확인
docker ps   

# (2) 컨테이너 중지 (b = 컨테이너 ID or 이름)
docker stop b   

# (3) 실행 중인 컨테이너가 있는지 다시 확인
docker ps -a  



# (1) 특정 컨테이너 삭제 (중지된 상태여야 함)
docker rm b    

# (2) 모든 컨테이너 삭제 (실행 중인 것도 포함, 강제)
docker rm $(docker ps -aq) -f  


# (1) 실행 중인 컨테이너 확인
docker ps  

# (2) 실행 중인 컨테이너 중지 (필수)
docker stop <container_id>   

# (3) 실행 중지 후 이미지 삭제 (-f: 실행 중인 것도 삭제)
docker rmi 이미지명 -f    

# (4) 사용하지 않는 이미지 전부 삭제 (dangling 이미지)
docker image prune -a

배포 방법 개요

• 컨테이너를 클라우드에 올린다고 바로 실행되지 않음

• 로컬에서는 Docker 컨테이너 실행만으로 API에 접근 가능
• 클라우드에서는
  • 컨테이너가 실행되더라도 네트워크 접근이 차단되어 있음
  • 로드 밸런서, 방화벽, 인바운드 규칙 설정이 필요할 수 있음
  • 게다가 여러 컨테이너를 운영하려면 오케스트레이션 필요
    • 보통 Kubernetes(쿠버네티스) 사용
    • 여러개를 관리하려면 필요한 개념

쿠버네티스

Kubernetes는 꼭 사용해야 할까?

• 단일 컨테이너라면 K8S 없이 실행 가능
• 하지만 트래픽 증가, 자동 스케일링, 무중단 배포, 2개 이상 컨테이너를 사용 시 쿠버네티스를 사용해야 함
• K8S 사용 시 배포 방법
  • 컨테이너를 Registry에 업로드
  • VPC, Subnet, NAT GW, Storage 설정
  • Kubernetes Service에 Registry 연결 (템플릿 배포)

Cloudtype 배포

1. Cloudtype 가입 (카드 등록 필수)
2. 컨테이너 기반 환경: SQLite 그대로 사용할 예정
   • 단 컨테이너가 재시작될 시 파일이 삭제될 수 있음 = 저장 경로를 지정함으로써 문제 해결

     database.py
     from sqlalchemy import create_engine, Column, Integer, String
     from sqlalchemy.orm import sessionmaker, declarative_base
     import os
     

DB_DIR = "./data" if not os.path.exists(DB_DIR): os.makedirs(DB_DIR)

DATABASE_URL = "sqlite:///./data/test.db"

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False}) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base()

class Ticket(Base): tablename = "tickets" id = Column(Integer, primary_key=True, autoincrement=True) name = Column(String, index=True) age = Column(Integer) price = Column(Integer)

Base.metadata.create_all(bind=engine)

def get_db(): db = SessionLocal() try: yield db finally: db.close()

3. cloudtype.yaml 추가
```yaml
# cloudtype.yaml
app: fastapi-tickets
service: web
env: python
start: uvicorn main:app --host 0.0.0.0 --port 8000

4. Github Repository 생성

5. Cloudtype 접속

6. Cloudtype 배포

Vercel 배포

• Serverless 환경
  • SQLite는 로컬 기반이며 Vercel에서 사용 불가
  • 따라서 외부 DB를 생성하여 연결하는 방향으로 진행 예정
• 배포 시 확인할 것
  • vercel.json
  • vercel CLI 설치 (npm 사전 설치 필요)
  • NeonDB, Vercel 회원가입
  • requirements.txt 더블체크 (특히 psycopg2-binary)

1. pip install psycopg2-binary + NeonDB로 접속 로그인 후 connect로 내 DB주소 확인, 아래 postgresql 부분 복사

2. database.py

   # database.py
   import os
   

DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://user:password@dbhost:5432/dbname")

engine = create_engine(DATABASE_URL)

3.vercel.json
```js
{
  "builds": [
    { "src": "main.py", "use": "@vercel/python" }
  ],
  "routes": [
    { "src": "/(.*)", "dest": "main.py" }
  ]
}

4. vercel 설치 및 프로젝트 초기화
   1. npm install -g vercel
   2. vercel login (vercel 웹사이트 회원가입)
      • Continue with GitHub(원하는 것에 맞춰)
      • vercel link
      • Scope 선택
      • N
      • 제작한 프로젝트 이름
      • ./
5. 환경변수 설정
   • vercel env add DATABASE_URL(postgres url 복붙)
   • Production
6. 배포 시작 : vercel .

7. 오류 발생시

8. 환경 오류 발생시

9. 완료

• 클라이언트와 서버 간 양방향 통신을 지원하는 프로토콜
• HTTP 요청과 다르게 한 번 연결되면 지속적으로 데이터 송수신 가능
  • 클라이언트가 요청하지 않아도 서버가 데이터를 보낼 수 있음 (Push 방식 지원)
• ws:// 또는 wss://(보안) 방식 활용(web socket security)
• 실시간 통신

필요 이유

= 기존 HTTP 방식의 문제점

• 실시간성 부족
  • HTTP 요청-응답 방식은 클라이언트가 요청해야만 서버가 응답 가능
  • 실시간 알림, 채팅 등에는 비효율적
• 풀링(Pulling)과 긴 연결(Long Polling)의 한계
  • 풀링: 클라이언트가 일정 시간마다 서버에 요청 (과부하 문제)
  • 긴 연결: 클라이언트가 연결을 유지하며 서버 응답을 기다림 (비효율적)

WebSocket 프로토콜 구조

pip3 install 'uvicorn[standard]'

• Handshake
  • HTTP 요청을 통해 WebSocket 연결 수립 (이 때 Upgrade: websocket헤더 사용)
  • 기존 HTTP 요청을 WebSOcket통신으로 전환하는 과정
  • 클라이언트와 서버간에 WebSocket 연결을 수립하기 위해 수행하는 초기 단계를 의미
  1. 클라이언트가 서버에 WebSocket 요청을 보냄 (Upgrade 요청
  2. 서버가 WebSocket 연결을 허용하면 응답 (101 Switching Protocols)
  3. WebSocket 연결이 수립되고, 지속적인 데이터 통신 가능
• 메시지 송수신 (Message Exchange)
  • 텍스트 또는 바이너리 메시지 교환
• 연결 종료(Connection Close)
  • 클라이언트 또는 서버가 연결 종료

🧑‍💻 서버 구현

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws") # 보편적으로 /ws로 url을 지정
async def websocket_endpoint(websocket: WebSocket): # 실시간으로 여러 요청을 처리하기에 비동기가 적합
    await websocket.accept() # WebSocket 연결 수락
    while True: 서버를 게속 유지하기위해 (돌리기 위해)
        data = await websocket.receive_text() # 클라이언트 메세지 수신
        await websocket.send_text(f"서버 응답: {data}") # 클라이언트에게 응답

• postman 설정

WebSocket 서버 구현 - Disconnect 1000

• Starlette.websockets.WebSocketDisconnect: (1000, ‘’)
  • 에러코드 1000: WebSocket이 정상적으로 종료됨
  • 클라이언트(예: Postman, 브라우저)가 WebSocket 연결을 닫으면 서버에서 발생하는 오류
• 해결 방법: try-except을 추가하여 예외 처리
  • fastapi에서 WebSocketDisconnect을 import 후, except 처리

from fastapi import FastAPI, WebSocket, WebSocketDisconnect


app = FastAPI()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"서버 응답 : {data}")
    except WebSocketDisconnect:
        print("클라이언트 연결 종료")

WebSocket을 활용한 실시간 기능

• 대표적인 서비스 사례
  • 실시간 채팅: 카카오톡, 슬랙, 디스코드 등
  • 실시간 알림 시스템: SNS 알림, 주문 처리 상태 알림 등
  • 실시간 주식 데이터 스트리밍: 증권 거래소 API
  • 실시간 협업 기능: Google Docs같은 실시간 편집

실시간 알림 시스템 구현

• 기본 개념

  • 클라이언트(Web, Mobile 등)가 서버와 WebSocket 연결을 맺음
  • 서버가 특정 이벤트(예: 주문 완료, 메시지 도착, 좋아요 등) 감지
  • 서버가 연결된 모든 클라이언트에게 즉시 알림 전송

• 시스템 구성도

• 여러 클라이언트가 WebSocket을 통해 서버에 연결될 수 있도록 구성

  • ConnectionManager 클래스를 만들어 연결 관리

  • 사용자가 연결을 끊으면 자동으로 제거됨

    class ConnectionManager:
    """WebSocket 연결관리"""
    def __init__(self): # 연결된 사용자 리스트
        self.active_connections: List[WebSocket] = []
    
    async def connect(self, websocket: WebSocket):
        """ 클라이언트가 websocket 연결을 요청하면 리스트에 추가 """
        await websocket.accept()
        self.active_connections.append(websocket)
    
    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
    
    async def broadcast(self, message: str):
        """ 모든 연결된 클라이언트에게 메세지 전송 """
        for connection in self.active_connections:
            await connection.send_text(message)
    

manager = ConnectionManager()

특정 요청을 감지하면 서버가 사용자에게 전달

* 특정 사용자에게만 1:1 알림 보내기
```python

from typing import Dict, List
from fastapi import FastAPI, WebSocket, WebSocketDisconnect


app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: Dict[str, WebSocket] = {}

    async def connect(self, websocket: WebSocket, username: str):
        """ 특정 사용자의 WebSocket 연결 관리 """
        await websocket.accept()
        self.active_connections[username] = websocket

    def disconnect(self, username: str):
        """ 사용자가 연결을 끊으면 제거 """
        if username in self.active_connections:
            del self.active_connections[username]

    async def send_private_message(self, username: str, message: str):
        """ 특정 사용자에게 메세지 전송 """
        if username in self.active_connections:
            await self.active_connections[username].send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{username}")
async def websocket_endpoint(username: str,websocket: WebSocket):
    """ 클라이언트가 /ws 경로로 WebSocket 연결 요청 """
    await manager.connect(websocket, username)

    try:
        while True:
            data = await websocket.receive_text()
            await manager.send_private_message(username, f"📢 개인 메시지: {data}")
    except WebSocketDisconnect:
        manager.disconnect(username)

@app.post("/send-message/{username}")
async def send_message(username: str, message: str):
    """ 특정 사용자에게 메시지를 전송하는 API (Postman 테스트용) """
    await manager.send_private_message(username, message)
    return {"message": f"📢 {username}에게 메시지 전송 완료!"}

• postman

WebSocket 보안 강화

• 보안 프로토콜 적용 (wss://)
• DDos 공격 방어
  • 디도스: 고의로 접속량을 폭주시켜 서버를 마비시키는 것
  • IP 기반 연결 제한
  • Rate Limiting: 일정 시간 내 특정 요청 횟수 초과 시 차단
  • Cloudflare, AWS WAF 같은 방화벽 솔루션 활용

WebSocket 성능 최적화

• WebSocket Keep-Alive 설정

  • 기본적으로 장시간 연결을 유지하므로 서버 부하 발생 가능
  • Ping/Pong 메시지를 사용하여 비활성 연결 종료
  • 특정 시간마다 클라이언트와 서버 간 메시지 송수신

    while Ture:
    await websocket.send_text("ping")
    await asyncio.sleep(30) # 30초마다 ping 전송

• 로드 밸런싱 Load Balancing

  • 하나의 서버가 많은 요청을 처리하면 서버 과부하 발생
  • 여러 개의 서버를 사용하여 트래픽을 분산
  • 클라이언트 요청을 여러 서버 중 하나로 자동으로 분배

• 사용자와 직접적인 상호작용 없이 실행되는 작업
• API 요청을 처리한 후 별도 스레드에서 작업 실행
• BackgroundTasks.add_tasks()를 사용하여 작업 추가

동기 vs 비동기 vs 백그라운드 작업 차이

• 백그라운드 작업 방식: 즉시 응답을 반환후, 실제 작업은 백그라운드에서 실행
  • 실행 예시 : 콘솔에선 5초후에 작업이 실행

    from fastapi import FastAPI, BackgroundTasks
    import time
    

app = FastAPI()

def background_task(): time.sleep(5) print("백그라운드 작업 완료")

@app.get("/") async def stsart_background_task(background_tasks: BacgroundTasks): background_tasks.add_task(background_task) return {"msg": "백그라운드 작업 실행 중..."}

### Background Tasks가 필요한 이유
> * 이메일 전송: 사용자가 요청하면 즉시 응답을 주고, 실제 이메일 전송은 백그라운드에서 진행
> * 파일 업로드: 파일을 업로드한 후, 처리는 백그라운드에서 수행
> * 로그 기록: API 호출 로그를 백그라운드에서 저장
> * 장기 실행 작업: 데이터 분석, PDF 생성, 크롤링 등

* 이점
  - **빠른 응답 시간**: 사용자는 즉시 응답을 받고 작업은 따로 실행
  - **서버 부하 감소**: 여러 요청이 들어와도 메인 스레드가 차단되지 않음
  - **유저 경험 개선**: 이메일 전송이나 데이터 저장 작업이 끝날 때까지 기다릴 필요 없음

### FastAPI의 Background Tasks 작동 방식

1. API 요청을 받음
2. add_task()에 해당 요청 처리
3. FastAPI는 일단 즉시 응답을 반환
4. add_task()에 대한 내용이 백그라운드에서 실행됨
5. 그 외 서버가 다른 요청을 처리하는 동안 작업이 진행됨

### Background Tasks 주의점
* 백그라운드 작업이 **실패**해도 클라이언트는 **모름**
  - 작업이 실패해도 응답은 이미 전송됨
  - 해결책: 작업 결과를 데이터베이스에 저장하고 후속 확인 API 제공

* 서버 **재시작** 시 작업이 **사라짐**
  - 서버가 재시작되면 미완료된 백그라운드 작업이 유실됨
  - 해결책: Celery같은 작업 큐 사용 고려

* 대량의 백그라운드 작업은 성능 저하 가능성
  - 한 번에 너무 많은 작업이 들어오면 서버 성능 저하
  - 해결책: 워커 프로세스를 활용하여 작업 분산

### 🧑‍💻 예시

* 이메일 전송 : 즉시 응답 주고, 이메일 전송은 백그라운드에서 진행
```python
from fastapi import FastAPI, BackgroundTasks 
import time
app = FastAPI()

def send email(email: str, message: str) :
    time.sleep(5)
    pirnt(f"이메일 전송 완료: {email}, 내용: {message}")
@app.post("/send-email/")
async def send_email_endpoint(email: str, message: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, email, message)
    return {"message": "이메일 전송 중..."}

• 로그 기록 : 요청이 끝난 후 로그가 저장 됨

  def write_log(log_message: str):
    with open("log.txt", "a") as log_file: # append 추가모드
    log_file.write(log_message + "\n")

• 사용자 활동 기록: DB에 사용자 활동을 백그라운드에서 기록

  # database.py
  from sqlalchemy import Column, Integer, String, create_engine 
  from sqlalchemy.orm import sessionmaker, declarative_base
  

DATABASE_URL = "sqlite:///./test.db" engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False}) # sqlite 전용 설정 SessionLocal = sessionmaker(autocommit=False, autoflush=False,bind=engine) Base = declarative_base()

class UserActivity(Base): tablename = "user_activities" id = Column (Integer, primary_key=True, index=True) username = Column (String, index=True) action = Column (String)

Base.metadata.create_all (bind=engine)

def get_db() : db = SessionLocal() try: yield db finally: db.close()

main.py

from fastapi import FastAPI, BackgroundTasks, Depends from sqlalchemy.orm import Session from database import get_db, UserActivity app = FastAPI()

def save_activity(username: str, action: str, db: Session): new_activity = UserActivity(username=username, action = action) db.add(new_activity) db.commit()

@app.post("/track-activity/") async def track_activity(username: str, action: str, background_tasks: BackgroundTasks, db: Session=Depends(get_db)): background_tasks.add_task(save_activity, username, action, db)

* backtracking 예시(파일 업로드)
```python
# database.py
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import Column, Integer, String, create_engine


DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

class FileRecord(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    filename = Column(String, index=True)
    status = Column(String)

Base.metadata.create_all(bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# main.py

from fastapi import BackgroundTasks, Depends, FastAPI, File, UploadFile
from pydantic import BaseModel
from sqlalchemy.orm import Session
import os

from database import get_db, FileRecord


app = FastAPI()

UPLOAD_DIR = "uploads"
# 지정된 경로의 폴더를 생성하는 함수
# 해당폴더가 이미 존재해도 에러발생 x
os.makedirs(UPLOAD_DIR, exist_ok=True)

# os.path.join(UPLOAD_DIR, filename)
# UPLOAD_DIR 디렉토리 안에 filename 파일을 저장할 경로를 만듭니다
# os.path.join()을 사용하면 OS에 따라 적절한 경로 구분자를 자동으로 적용합니다.

# "wb" 모드는 바이너리 쓰기 모드
# 텍스트 파일이 아닌 이미지, 동영상, PDF 같은 바이너리 파일을 저장할 때 사용
def save_file(filename: str, content: bytes):
    with open(os.path.join(UPLOAD_DIR, filename), "wb") as f:
        f.write(content)
    print(f"파일 저장 완료: {filename}")

# ValueError: I/O operation on closed file. 발생
# 로컬에서 파일으 업로드하기에 경로를 명시적으로 지정을 안하면 인식을 못함, 아래 비추
# def save_fiel(file: UploadFile):
#     with open(f"uploads/{file.filename}", "wb") as f:
#         f.write(file.file.read())
#     pritn(f"파일 저장 완료: {file.filename}")

def log_file_upload(filename: str, db: Session):
    new_record = FileRecord(filename=filename, status="Saved")
    db.add(new_record)
    db.commit()
    print(f"로그 저장 완료: {filename}")

@app.post("/upload-file/")
async def upload_file(file: UploadFile = File(...),
                background_tasks: BacgroundTasks = BackgroundTasks(),
                db: Session = Depends(get_db)):
    file_contet = await file.read() # 파일을 미리 읽어서 백그라운드 작업으로 전달
    background_tasks.add_task(save_file,file.filename, file_content)
    bacgroound_tasks.add_task(log_file_upload, file.filename, db)

    return {"msg" : " 파일 업로드 중 ..."}

• DB에 넣으려면, 테이블에 추가 (LargeBinary)

  from fastapi import FastAPI, UploadFile, File, HTTPException
  from sqlalchemy import Column, Integer, LargeBinary, create_engine
  from sqlalchemy.ext.declarative import declarative_base
  from sqlalchemy.orm import sessionmaker, Session
  

app = FastAPI()

Base = declarative_base() DATABASE_URL = "sqlite:///./test.db" engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False}) SessionLocal = sessionmaker(bind=engine)

class FileData(Base): tablename = "files" id = Column(Integer, primary_key=True, autoincrement=True) data = Column(LargeBinary)

Base.metadata.create_all(bind=engine)

@app.post("/upload/") async def upload_file(file: UploadFile = File(...)): db: Session = SessionLocal() file_content = await file.read()

new_file = FileData(data=file_content)
db.add(new_file)
db.commit()
db.refresh(new_file)
db.close()

return {"file_id": new_file.id}

@app.get("/download/{file_id}") async def download_file(file_id: int): db: Session = SessionLocal() file_data = db.query(FileData).filter(FileData.id == file_id).first() db.close()

if not file_data:
    raise HTTPException(status_code=404, detail="File not found")

return {"file_content": file_data.data}  # 실제 서비스에서는 파일을 반환하는 방식 사용

* api 코드는 동일함 save 할때 그냥 contet를 추가적으로 넣으면됨
```python
new_record = FileRecord(filename=filename, status="Saved", file = content)

• DB에서 이미지 불러오기, 똑같이 가지고 오면되지만 return값에, media_type ="png/image"를 넣어준다

  @app.get("/image/{image_id}")
  async def get_image(image_id: int):
    db: Session = SessionLocal()
    image_data = db.query(ImageData).filter(ImageData.id == image_id).first()
    db.close()
  
    if not image_data:
        raise HTTPException(status_code=404, detail="Image not found")
  
    return Response(content=image_data.data, media_type="image/png")

테스트가 필요한 이유

• 코드 변경 시 기존 기능이 정상적으로 작동하는지 보장하기 위해
• 자동화된 테스트를 통해 버그를 빠르게 발견하고 수정 가능
• 특히 API 개발에서는 요청/응답을 검증하는 것이 중요

1. 유닛테스트
2. 모델 테스트
3. 라우터 테스트
4. 통합 테스트

pytest

Python의 대표적인 테스트 프레임워크

• pip install pytest

httpx

FastAPI에서 requests 대신 사용하는 비동기 HTTP 클라이언트

• pip install httpx

TestClient

FastAPI 애플리케이션에 직접 요청을 보내고 응답 검증

종류

• 단위 테스트(Unit Test) : 개발 함수, 모듈 단위로 동작을 검증
• 통합 테스트(Integration Test) : 여러 모듈이 함꼐 동작할 때의 결과를 검증
• 엔드투엔드 테스트(E2E Test) : 실제 사용자 시나리오를 기반으로 전체 시스템을 테스트

환경구성

• pytest 및 httpx 패키지 설치 : pip install pytest httpx
• FastAPI 설정: 테스트 하고싶은 엔드포인트 작성 : 파일명 main.py
• 테스트 코드 실행, pytest test_main.py

• 기본적인 pytest 테스트 코드 작성
  • test_main.py

    from fastapi.testclient import TestClient
    from main import app
    

client = TestClient(app)

def test_ping(): response = client.get("/ping") assert response.status_code == 200 assert response.json() == {"msg": "pong"}

pytest test_main.py

### 테스트 시나리오

#### 존재하지 않는 엔드포인트 테스트
```python
def test_not_found():
    response = client.get("/invalid-endpoint")
    assert response.staus_code == 404

필수 파라미터가 없을때 예외 처리

@app.get("/greet/")
async def greet(name: str):
    return {"msg" : f"Hello, {name}!"}

def test_greet():
    response = client.get("/greet/", params = {"name" : "Alice"})
    assert response.status_code == 200
    assert response.json() == {"msg" : "Hello, Alice!"}

def test_greet_missing_param():
    response = client.get("/greet/")
    assert response.status_code == 422

단위 테스트 (Unit Test)

• 단위 테스트란?

  • 코드의 개별 기능(함수, 클래스, 모듈) 단위로 동작을 검증하는 테스트
  • 데이터베이스, 외부 API 호출 등 의존성이 없는 상태에서 실행됨

• 단위 테스트의 필요성

  • 코드 변경 후에도 기능이 정상적으로 유지되는지 보장
  • 버그를 조기에 발견하여 수정 비용 절감
  • 빠른 실행 속도로 자주 실행 가능

• 최대한 많은 케이스를 찾아서 테스트

from fastapi import FastAPI
    app = FastAPI ()
def subtract(a: int, b: int):
    # 빼기 연산 함수
    return {" result": a - b}
@app.get("/math/subtract")
def subtract_route(a: int, b: int):
# FastAPI 엔드포인트 (subtract API)
    return subtract(a, b)

● 정상적인 뺄셈
● 음수 포함 뺄셈
● 0을 포함한 뺄셈
● 잘못된 입력값
● 경계값 (최소, 최대값)

Depends를 활용한 단위 테스트

app = FastAPI()

def get_default_number():
    # 의존성 함수
    return 10

def subtract(a: int, b:int, default= int = Depends(get_default_number)):
    return {"result" : a-b-default}

@app.get("/math/subtract")
async def subtract_endpoint(a: int, b:int, default= int = Depends(get_default_number)):
    return subtract(a, b, default)

• 의존성 주입 함수에 dependency_overrides로 테스트 데이터를 덮어씌움

  통합 테스트 (Integration Test)

  • 통합 테스트의 필요성
    • 단위 테스트만으로는 모듈 간 연결이 올바르게 동작하는지 확인할 수 없음
    • API와 데이터베이스 간의 데이터 흐름이 정상적으로 이루어지는지 검증 필요

  • 다양한 API 요청 조합을 테스트하여 전체 시스템이 예상대로 동작하는지 확인

# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel
from database import get_db, Product

app = FastAPI()

# 요청 데이터 모델 정의
class ProductCreate(BaseModel):
    name: str
    price: int

# 제품 추가 API
@app.post("/products/")
async def create_product(product: ProductCreate, db: Session = Depends(get_db)):
    product = Product(name=product.name, price=product.price)
    db.add(product)
    db.commit()
    db.refresh(product)
    return product

# 제품 조회 API
@app.get("/products/{product_id}")
async def read_product(product_id: int, db: Session = Depends(get_db)):
    product = db.query(Product).filter(Product.id == product_id).first()
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    return product

# database.py
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class Product(Base):
    __tablename__ = "products"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    price = Column(Integer)

Base.metadata.create_all(bind=engine)

# DB 세션 의존성
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# test_main.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from fastapi.testclient import TestClient
from database import Base, get_db, Product
from main import app

# 테스트용 데이터베이스 설정
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

client = TestClient(app)

# 한 번만 실행하여 전체 테스트에서 DB를 공유
@pytest.fixture(scope="module")
def setup_test_db():
    Base.metadata.create_all(bind=engine)  # 새로 생성
    db = TestingSessionLocal()

    # 초기 데이터 삽입
    test_product = Product(id=1, name="Laptop", price=1500)
    db.add(test_product)
    db.commit()

    yield db  # 모든 테스트에서 공유됨

    db.close()
    Base.metadata.drop_all(bind=engine)  # 모든 테스트 종료 후 DB 삭제

# 제품 추가 API 테스트
def test_create_product(setup_test_db):
    response = client.post("/products/", json={"name": "Phone", "price": 800})
    assert response.status_code == 200
    assert response.json()["name"] == "Phone"
    assert response.json()["price"] == 800

# 존재하는 제품 조회 테스트 (Laptop & Phone 조회 가능)
def test_read_product(setup_test_db):
    response = client.get("/products/1")  # ID=1 Laptop 조회
    assert response.status_code == 200
    assert response.json()["name"] == "Laptop"
    assert response.json()["price"] == 1500

    response = client.get("/products/2")  # ID=2 Phone 조회
    assert response.status_code == 200
    assert response.json()["name"] == "Phone"
    assert response.json()["price"] == 800

# 존재하지 않는 제품 조회 테스트 (404 반환)
def test_read_product_not_found(setup_test_db):
    response = client.get("/products/999")  # 존재하지 않는 ID 요청
    assert response.status_code == 404
    assert response.json()["detail"] == "Product not found"

테스트 코드 최적화 개념

• 테스트 코드가 많아질수록 발생하는 문제
  • 실행 속도가 느려짐
  • 중복 코드가 많아짐
  • 테스트 케이스가 늘어남

테스트 최적화를 위한 핵심 기법

Mocking

• Mocking: DB나 외부 API 호출 없이 가짜 데이터를 반환하도록 처리
  • 실제 데이터베이스와의 연결 없이 API 로직만 검증 가능
  • 테스트 시 불필요한 DB 접근을 막고, 실행 속도를 개선
• 컴퓨터 입장에서 실제 DB를 연결하지 않고 뒤에 값이 저걸 쓰려나 보다 라고 생각을함
• 그래서 부하가 안걸림, 기존의 사용에 .return_value 만 붙여주면 됨

• patch 에서 앞의 main에 불러와진 모듈은 전부 사용가능 그래서 get_db가 import 되어있어서 사용 가능

  • @patch("모듈.함수") 에서 "함수"부분이 mock_함수명 형식으로 매칭
  • 여러개의 @patch를 사용할 경우, 가장 마지막 @patch가 첫번쨰 매개변수로 등록
  • Mocking된 객체는 MagicMokc() 타입이므로, .return_value를 설정해서 동작을 정의할 수 있음
  • 정의된 순서대로 아래부터 바로 적용할 함수에 순차적으로 적어야 함

    @patch("auth.verify_password")
    @patch("auth.create_token")
    def test_login(mock_create_token, mock_verify_password, mock_db, username, password, expected_status):

• 기존방식

  @app.get("/users/(user_id}")
  async def get_user(user_id: int, db: Session = Depends (get_db)):
    user = db.query(User).filter(User.id = user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user
  

def test_get_user(): response = client.get("/users/1") assert response.status_code ==200

* **Mocking 활용**
```python
@app.get("/users/(user_id}")
async def get_user(user_id: int, db: Session = Depends (get_db)):
    user = db.query(User).filter(User.id = user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

# mocking
from unittest.comk import MacigMock

def test_get_user_with_mock():
    mock_db = MagicMock()
    mock_db.query.return_value.filter.return_value.first.return_value= User(id=1, name="Mock User", email = "mock@example.com")

    response = client.get("/users/1", dependencies={"db": mock_db})
    assert response.status_code == 200
    assert response.json()["name"] == "Mock User"

# patch
@patch("main.get_db") # 데이터 베이스가 있는거 같다 그냥 생각을 하게 만듬 컴퓨터에(실제 DB가 없어도 되는거임)
def test_get_user_with_mock(mock_get_db):
    mock_db = MagicMock()
    mock_db.query.return_value.filter.return_value.first.return_value =User(id=1, name="Mock User", email ="mock@example.com")
    # Patch를 적용하면서 FastAPI의 의존성 오버라이드 설정
    mock_get_db.return_value = mock_db
    app.dependency_overrides[get_db] = lambda: mock_db

    response = client.get("/users/1")

    assert response.status_code == 200
    assert response.json()["name"] == "Mock User"

    # 테스트 완료 후 의존성 초기화
    app.dependency_overrides.clear()

• AsyncMock

  • API 최적화를 위해 사용 할 경우도 있음 대부분 MagicMock 으로 처리

from unittest.mock import AsyncMock

mock_db = AsyncMock()  
mock_db.query.return_value.filter.return_value.first.return_value = User(id=1, name="Async Mock User", email="asycn@example.com")

Fixture

• Fixture: 테스트 실행 전 공통 환경을 설정하는 기능
  • 여러 테스트에서 중복된 코드 없이 동일한 설정 공유
  • DB 초기화, API 클라이언트 생성, 기본 데이터 삽입 등에 사용
• 공통 환경을 설정하는 기능

• Fixture의 scope 설정

• function : 각각의 함수에 각각의 데이터베이스를 세팅

  • 한 함수가 끝나면 그 데이터베이스가 초기화 됨

• module : 중복된 데이터를 줄이면서 테스트 하기위해, 한 파일안에서 중복된 데이터 사용

• 예시

  def test_create_user():
    db = TestingSessionLocal()
    Base.metadata.create_all(bind=engine)
    ...
    Base.metadata.drop_all(bind_engine)
    db.close()
  

변경

import pytest

@pytest.fixture(scope="module") def test_db(): """ 한 번만 실행하여 전체 테스트에서 공유하는 데이터베이스 설정 """ Base.metadata.drop_all(bind=engine) Base.metadata.create_all(bind=engine) db = TestingSessionLocal() yield db # 모든 테스트에서 공유됨 db.close() Base.metadata.drop_all(bind=engine) # 모든 테스트 종료 후 DB 삭제

### Parametrize
> * Parametrize: 동일한 테스트를 다양한 입력값으로 실행할 때 사용
>   - @pytest.mark.parametrize를 활용하여 자동 반복 실행
>   - 코드 중복 없이 테스트를 간결하게 유지 가능

* 데코레이터 밑의 인덱스로 처리된 값들을 데코레이터 안의 파라미터에 맞춰 데이터들을 순차적으로 전부 테스트함

* 기존방식
```python
@app.get("/math/add")
async def add_numbers(a: int, b: int):
    return {"result" : a + b}

def test_add_numbers_1():
    response = client.get("/math/add?a=3&b=5")
    assert response.status_code == 200
    assert response.json()["result"] == 8
def test_add_numbers_2():
    response = client.get("/math/add?a=-2&b=7")
    assert response.status_code == 200
    assert response.json()["result"] == 5

• Parametrize 적용

  import pytest
  @pytest.mark.parametrize("a, b, expected", [
    (3, 5, 8),
    (-2, 7, 5),
    (10, -10, 0),
    (0, 0, 0)
  ])
  def test_add_numbers(a, b, expected):
    response = client.get(f"/math/add?a={a}&b={b}")
    assert response.status_code == 200
    assert response.json()["result"] == expected
  

@app.post("/users/") async def create_user(user: dict): return {"id": 1, "name": user["name"], "email": user["email"]}

@pytest.mark.parametrize("user_data, expected_status", [ ({"name": "Alice", "email": "alice@example.com"}, 200), ({"name": "Bob"}, 422), # 이메일 누락 ({"email": "invalid.com"}, 422), # 이름 누락 및 이메일 형식 오류 ]) def test_create_user(user_daa, expected_status): response = client.post("/users/", josn=user_data) assert response.status_code == expected_status

### 성능 측정
> * 테스트 실행 시간이 길어질 경우, 특정 API의 성능을 분석해야 함
>   - time.time()을 사용하여 API 응답 시간을 측정
>   - 특정 API가 성능 저하를 일으키는지 파악 가능\

```python
import time

def test_api_performance():
    start_time = time.time()
    response = client.get("/math/add?a=10000&b=20000")
    end_time = time.time()

    assert response.status_code == 200
    assert response.json()["result"] == 30000
    print(f"API 실행 시간: {end_time -start_time:.5f}초"}

Tip

• email : EmailStr | None = None
  • EmailStr 이메일 형식 검증, pydnatic 모듈에 있음
  • Or none 선택적이라고 명시하는것 Optional 과 같음 3.10 버전 이후부터 존재
  • EamilStr = None 이랑 같음

✅ 1. Pydantic vs Django Serializer 비교

🔹 결론:

• Pydantic은 DB 모델과 완전히 독립적이며, 요청 데이터를 검증하는 데 사용됨.
• Django Serializer는 Django 모델과 연동 가능 (ModelSerializer 지원)

pydnatic의 validator 개념 및 사용법

• 구조
  • @field_validator("<필드 이름>"), @classmethod 데코레이터를 사용하여 특정 필드의 값을 검증
  • 검증 함수는 항상 클래스 메서드 형태로 정의되며, 첫 번째 인자로 필드값을 받음
  • 검증 실패 시 예외를 발생시켜야 함 : ValueError, TypeError...
• 예시

  from pydantic import BaseModel, field_validator
  

class ExampleModel(BaseModel): number: int

@field_validator("number")
@classmethod
def validator_number(cls, value):
    if value < 1 :
        raise ValueError("Number must be at least 1")
    if value > 100 :
        raise ValueError("Number must not exceed 100")
    return Value

* 동작
  - 모델 초기화 시, 해당 필드의 값이 검증 로직을 통과해야 함
  - 통과하지 못하면 예외 메세지가 반환
```python
example = ExampleModel(number = 50)
print(example) # Output : number 50

• 데이터 검증이 필요한 이유
  • 사용자가 입력하는 데이터가 항상 올바르지는 않음
  • 보안문제 : 잘못 입력된 SQL injection, XSS등 보안 취약점을 유발할 수 있음
  • 비즈니스 로직 보장 : 이메일 형식, 전화번호 국가별 형식, 나이 제한 등

고급 pydantic 검증

• 단일 필드 검증을 넘어 필드 간 관계를 고려한 검증이 필요
• 실제 서비스에서 발생할 수 있는 논리적 오류를 방지
• 보안, 데이터 정합성을 보장하기 위해 더 정교한 검증이 필요

@field_validator

• 비밀번호 검증

  from pydantic import BaseModel, field_validator, ValidationError
  

special_word = "!@#$%^&*()-_+="

class UserRegister(BaseModel): username : str password : str

@field_validator("password")
@classmethod
def validate_password(cls, value):
    if len(value) < 8:
        raise ValueError("password must be at least 8 characters")
    flag = False
    for char in value:
        if char.isupper():
            flag = True
            break
    if count == 0 :
        raise ValueError("password must contain at least one uppercase letter")

    if not any(char in special_word for char in value):
        raise ValueError("password must contain at least one special character")

    return value

* re 정규식 사용
```python
class UserRegister(BaseModel):
    username: str
    password: str

    @field_validator("password")
    @classmethod
    def validate_password(cls, value):
        if len(value) < 8:
            raise ValueError("Password must be at least 8 characters long")
        if not re.search(r"[A-Z]", value):
            raise ValueError("Password must contain at least one uppercase letter")
        if not re.search(r"\d", value):
            raise ValueError("Password must contain at least one number")
        if not re.search(r"[!@#$%^&*(),.?\":{}|<>]", value):
            raise ValueError("Password must contain at least one special character")
        return value

mode, @model_validator

poetry add email-validator

• before

  • 시점: 데이터 유효성 검증 전에 실행(데이터 받기전에 변환시키고)
  • 데이터 접근 방식: dict 형태로 원시 데이터 접근
  • 예제: 입력데이터 변환(소문자변환, 공백 제거, ...), 타입변환(str -> int)

• after

  • 시점: 데이터 유효성 검증 후 실행(데이터 받고 검사)
  • 데이터 접근 방식: self 객체를 사용해 필드값 접근
  • 예제: 여러 필드간 검증, 값의 논리적 검증, 비즈니스 로직 반영(role 기반 접근 제한)

class UserRegister(BaseModel):
    username : str
    password : str
    confirm_password : str

    @model_validator(mode="after")
    def check_password_match(self):
        if self.password != self.confirm_password:
            raise ValueError("Passwords do not match")
        return self

from pydantic import BaseModel, model_validator, Field, EmailStr

class ContactInfo(BaseModel):
    email : EmailStr | None = None
    phone_number : str | None = None # Optional , 선택적이란 의미

    @model_validator(mode="after")
    def exist_email_phone(self):
        if self.email is None and self.phone_number is None:
            raise ValueError("Must have email or phone_number")
        return self


# no_email = ContactInfo(phone_number="phone_number")
no_phone_number = ContactInfo(email="email@email.com")
# no = ContactInfo()
print(no_phone_number.email)

    @model_validator(mode="before")
    @classmethod
    def preprocess_username(cls, data):
        #  객체가 특정 클래스(또는 데이터 타입)의 인스턴스인지 확인하는 함수
        if isinstance(data, dict) and "username" in data:
            data["username"] = data["username"].lower()
        return data

@computed_field

• 자동계산 필드
  • 입렵값 기반으로 자동 계산되는 필드
    • 나이대신 생년원일을 입력받고 자동으로 계산
    • 정가와 할인율을 입력하면 할인가를 자동으로 계산

from pydantic import BaseModel, computed_field
from datetime import datetime

class User(BaseModel):
    name : str
    birth_year : int

    @computed_field
    @property
    def age(self) -> int:
    return datetime.now().yaer - self.birth_year

user = User(name="Alice", brith_year = 2000)
print(user.age) # output : 25

초기값 동적 설정, default_factory

• 초기값을 동적으로 설정할 떄 사용
  • 자동증가 ID
  • created_at을 현재시간으로 설정
  • verification code를 위해 6개의 랜덤숫자 설정
• default_factory는 해당 필드에 값이 제공되지 않았을 때, 기본적으로 어떤 값을 넣을지를 지정하는 기능
• 기본값을 제공하지 않으면 실행되는 기능

from pydantic import BaseModel, Field
from datetime import datetime

class LogEntry(BaseModel):
    message: str
    created_at: datetime = Field(default_factory=datetime.utcnow)

log1 = LogEntry(message="System started")
print(log1.created_at) # 자동생성된 UTC 시간

# 
import uuid
from datetime import datetime
from pydantic import BaseModel, Field

class User(BaseModel):
    # uuid.uuid4() : 랜덤한 UUID 생성
    user_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    name: str
    role: str = Field(default="user")
    created_at : str = Field(default_factory=lambda: datetime.now().strftime("%Y-%m-%d %H:%M:%S"))

user1 = User(name="alice")
print(user1)

#
import random
from datetime import datetime, timedelta
from pydantic import BaseModel, Field, field_serializer

class Otp(BaseModel):
    phone_number : str
    otp: int=Field(default_factory=lambda: random.randint(100000,999999))
    otp_expiry: str=Field(default_factory=lambda : (datetime.now()+ timedelta(minutes=5)).isoformat())

    # @field_serializer("otp_expiry")
    # def serialize_datetime(self, value: datetime) -> str:
    #     return value.isoformat()


otp = Otp(phone_number="010")

print(otp)

여러타입 지원

• 특정 필드에 여러 타입 지원
  • product_id가 int또는 str일 경우
  • score가 int또는 float일 경우
• Union 사용

from pydantic import
from typing import Union

class Product(BaseModel):
    product_id: Union[int, str]
    name: str

p1 = Product(product_id=123, name="Laptop")
p2 = Product(product_id="XYZ-456", name="Phone")

print(p1.product_id) # 123
print(p2.product_id) # "XYZ-456"

직렬화, json_encoders, @field_serializer

• 직렬화 Serialization: 다양한 종류의 데이터를 기계가 쓰고 읽기 편리하게 나타낸 방식
  • 기본적으로 datetime, Decimal 등은 JSON으로 직렬화할 수 없음
  • API응답에서 날짜 형식을 통일하거나 소수점 자리수 제한할 때 필요
• json_encoders는 Pydantic에서 특정 데이터 타입을 JSON으로 변환할 때 사용하는 커스텀 인코더
• datetime을 직접 str 형식으로 변환해서 JSON으로 출력하고 싶을 때 사용
• 기본 FastAPI 서버내에서는 기본값 유지

from pydantic import BaseModel
from datetime import datetime

class Order(BaseModel):
    order_id: int
    total_price: float
    created_at: datetime

    # class Config:
    #     json_encoders = {
    #         datetime: lambda v: v.strftime("%Y-%m-%d %H:%M:%S")
    #     }

    @field_serializer("created_at")
    def serialize_datetime(self, value: datetime) -> str:
        return value.strftime("%Y-%m-%d %H:%M:%S")

order = Order(order_id=1, total_price=100.5, created_at = datetime.now())
print(order.model_dump_json())

응답 필터링, response_model_exclude

• 응답에서 민감한정보(비밀번호, API Key 등)을 숨기고 싶을 떄
• 유저 권한에 따라 보여줄 데이터 조절
  • admin은 모든 필드
  • 일반 유저는 일부 피륻

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    username: str
    email: str
    password: str

@app.get("/users/me", response_model=User, response_model_exclude={"password"})
async def get_user():
    user_data = {"username":"testuser","email":"test@test.com","password":"mypassword"}
    return User(**user_data)

# username, email 만 출력됨

JSON & XML 응답 포맷 변환, Response, xml.etree.ElementTree

• API가 JSON 뿐만 아니라 XML도 지원해야할 때
• 사용자가 원하는 응답 포맷을 동적으로 변경하도록 설정

from fastapi import FastAPI
from fastapi.responses import JSONResponse, Response
import xml.etree.ElementTree as ET

app = FastAPI()

@app.get("/data/")
async def get_data(format: str="json"):
    data = {"message": "Hello, FastAPI"}

    if format == "xml":
        root = ET.Element("response")
        message = ET.SubElement(root, "message")
        message.text = data["message"]
        xml_str = ET.tostring(root, encoding="utf-8", method="xml")
        return Response(content=xml_str, media_type="application/xml")

    return JSONResponse(content=data)

# GET /data/?fromat=xml
# <response>
#     <message>Hello, FastAPI</message>
# </response>

다국어 응답처리(Accept-Language)

• 글로벌 API에서는 사용자의 언어에 따라 다른 메세지 반환 필요
• 브라우저 요청 헤더 Accept-Lanugage를 사용하여 자동 감지
• 지원하는 언어가 없으면 기본값을 설정해야 함

from fastapi import FastAPI, Header

app = FastAPI()

    responses = {
        "en": {"message": "Hello, welcome!"},
        "ko": {"message": "안녕하세요, 환영합니다!"},
        "fr": {"message": "Bonjour, bienvenue!"}
    }
@app.get("/greet/")
async def greet(accept_language: str = Header("en")):
    return responses.get(accept_language, responses["en"]) # 기본값 en

• 인증이 중요한 이유?
  • 보안 강화: 사용자 데이터 및 시스템 자원을 보호
  • 권한 관리: 인증을 통해 사용자의 권한을 검증
  • 사용자 경험: 안전한 환경에서 맞춤형 서비스 제공

인증 vs 인가

• 인증 : 너가 누구냐
• 인가 : 인증이 됬으면 너가 뭘 할수있냐

JWT(Json Web Token)

• 인증 정보를 JSON형식으로 저장하고, 서명을 통해 위변조 방지

• 주로 클라이언트와 서버간의 인증을 위해 사용

• HEADER/PAYLOAD/SIGNATURE

• 장점

  • 무상태 인증: 서버가 세션을 유지할 필요 없음
  • 확장성: 다양한 서비스에서 쉽게 통합 가능

• 단점

  • 크기가 크므로 반복적으로 보내는 경우 성능 이슈 발생
  • 만료되지 않은 토큰이 탈취되면 보안 문제가 발생

인증 디펜던시 (Depends 개념)

• 인증 디펜던시란?
  • FastAPI의 Depends를 사용하면 특정 기능 (예: DB 연결, 인증 등)을 경로 함수에서 반복적으로 재사용할 수 있음
• 인증 디펜던시가 필요한 이유
  • 인증 및 사용자 검증 로직을 코드 중복 없이 모듈화할 수 있음
  • 인증이 필요한 API 엔드포인트에서 간편하게 적용 가능
  • JWT 토큰을 자동으로 검증하고 사용자 정보를 API 경로 함수에 전달할 수 있음

인증 디펜던시의 동작 방식

• 기본적인 인증 흐름

  1. 클라이언트가 로그인하여 JWT 토큰을 발급받음
  2. 클라이언트는 모든 요청의 Authorization 헤더에 JWT 토큰을 포함하여 보냄
  3. FastAPI는 Depends(get_current_user)를 통해 인증 디펜던시를 실행
  4. JWT 토큰 검증 과정 a. 토큰이 유효한지 확인: jwt.decode b. 만료 여부 확인 c. 사용자가 존재하는지 데이터베이스에서 확인
  5. 인증 성공 시, 해당 사용자 정보를 경로 함수에 전달

• JWT 설정: Oauth2PasswordBearer(tokenUrl=“login”)

  • 클라이언트가 Authorization: Bearer <jWT_TOKEN> 형태로 요청을 보낼 때 추출하는 역할
  • tokenUrl=“login”은 로그인 API 엔드포인트를 지정하는 용도

인증 디펜던시 구현 방법

• 인증이 필요한 엔드포인트 적용
  • ex) 프로필 조회 API

    from fastapi.security import OAuth2PasswordBearer
    from jose import jwt, JWTError
    from passlib.context import CryptContext
    

class Token(BaseModel): access_token : str token_type : str

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

Passlib 라이브러리를 사용해 비밀번호를 안전하게 해싱하고 검증하기위한 설정

bcrypt 알고리즘을 사용하여 비밀번호 암호화

deprecated 사용해 이전 해싱 방식이 자동으로 업데이트 되도록 설정

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

OAuth2 인증 방식 중 “Password Flow” 방식을 사용하여 액세스 토큰을 발급하고 인증을 수행

OAuth2PasswordBearer(tokenUrl="login") → 사용자가 로그인할 때 JWT 토큰을 발급하는 URL을 지정

FastAPI의 Depends(oauth2_scheme)를 통해 자동으로 요청 헤더에서 토큰을 추출 가능

def hash_password(password: str): return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str): return pwd_context.verify(plain_password, hashed_password)

SECRET_KEY = "secret-key" ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict): # JWT 엑세스 토큰을 생성하는 함수 to_encode = data.copy() # data 딕셔너리를 복사하여 변경해도 원본 데이터가 영향을 받지 않도록 보호 expire = datetime.now() + timedelta(minutes = ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire}) # to_encode 딕셔너리에 "exp"(만료 시간) 키 추가 # JWT를 생성할 때 만료 시간이 포함되어야 자동으로 검증 가능 encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) # JWT를 생성하는 핵심 코드 return encoded_jwt

def get_current_user(token: str = Depends(oauth2_scheme)): """ 1. 전달된 JWT 토큰을 검증하여 사용자 정보를 가져오는 함수입니다. 2. JWT 토큰이 유효한지 확인하고, 만료 여부를 체크해야 합니다. 3. 토큰에서 사용자 정보를 추출한 후, 데이터베이스(또는 가짜 DB)에서 해당 사용자가 존재하는지 확인해야 합니다. 4. 사용자가 없거나 토큰이 유효하지 않다면, 401 Unauthorized 오류를 반환해야 합니다. """ try: # 1. JWT 토큰 디코딩 # SECRET_KEY와 알고리즘을 사용하여 토큰을 디코딩합니다. payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])

    # 2. 사용자 이름 추출
    # 토큰 페이로드에서 "sub" (subject), "exp" 필드를 가져옵니다.
    username = payload.get("sub")
    exp = payload.get("exp")

    # 3. 토큰이 유효하지 않거나 sub 필드가 없으면 오류 반환
    if username is None:
        raise HTTPException(status_code=401, detail="Invalid token")

    # 4. 사용자 검증
    # 데이터베이스(또는 가짜 DB)에서 사용자가 존재하는지 확인합니다.
    if username not in fake_users_db:
        raise HTTPException(status_code=401, detail="User not found")

    # 5. 토큰 만료 시간 검증
    # exp시간을 확인합니다.
    if exp is None or datetime.utcnow() > datetime.utcfromtimestamp(exp):
        raise HTTPException(status_code=401, detail="Token has expired")

    # 6. 검증된 사용자 정보 반환
    return fake_users_db[username]

except JWTError:
    # 7. 토큰 검증 실패 시 오류 반환
    raise HTTPException(status_code=401, detail="Invalid token")

Login API

@app.post("/login") async def login_user(user: UserLoginRegister, db: AsyncSession= Depends(get_db)): result = await db.execute(select(User).where(User.username == user.username)) existing_user = result.scalars().first() if not existing_user or not verify_password(user.password, existing_user.password): raise HTTPException( status_code=404, detail="User not found" ) access_token = create_access_token(data={"sub": existing_user.username}) return {"access_token": access_token, "token_type": "bearer"}

Profile API

@app.get("/profile") def get_profile(current_user: dict = Depends(get_current_user)): return {"username": current_user["username"]}

#### 인증 디펜던시 장점
* 코드 재사용성 증가
  - Depends(get_current_user)를 활용하면 API마다 중복된 인증 코드 작성이
불필요
  - 여러 API에서 일관된 방식으로 인증 적용 가능
* 보안성 강화
  - JWT 토큰을 중앙에서 검증: 일괄적인 보안 적용 가능
  - 모든 요청마다 get_current_user를 호출하여 인증을 검증
* 유지보수 용이
  - 인증 방식이 변경되더라도, get_current_user만 수정하면 전체 API에 반영됨
  - 관리가 쉽고, 확장성이 높음

### 🧑‍💻 DB에 연결한 모든 코드

```python
from datetime import datetime, timedelta, timezone

from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError

from pydantic import BaseModel
from passlib.context import CryptContext
from sqlalchemy import Column, String, Integer, select
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base



DATABASE_URL = "sqlite+aiosqlite:///./test.db"

async_engine = create_async_engine(
    DATABASE_URL,
    echo=True, # 쿼리 로그 출력
)

AsyncSessionLocal = sessionmaker(
    bind=async_engine, # 어떤 DB에 연결할지 설정
    class_=AsyncSession, 
    expire_on_commit=False # 세션이 커밋될 때 객체가 사라지지 않도록 설정
    # 📌 기본적으로 SQLAlchemy는 세션이 commit()되면 객체를 메모리에서 제거
    # 📌 expire_on_commit=False를 설정하면 세션이 종료되더라도 객체를 계속 유지 가능
)

Base = declarative_base()
# Base: 모든 데이터베이스의 부모 역할
# SQLAlchemy ORM(Object Relational Mapper) 모델을 정의하는 기본 클래스를 생성
# 📌 SQLAlchemy에서 ORM을 사용할 때, 모든 모델 클래스는 Base를 상속받아야 함.
# 📌 declarative_base()는 테이블을 정의할 때 필요한 “메타데이터”를 포함하는 베이스 클래스를 제공.


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String)
    password = Column(String)

app = FastAPI()

# DB 초기화
async def init_db():
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
        # 데이터베이스가 아직 존재하지 않는다면 테이블을 생성
        # 이미 존재하는 경우, 기존 테이블을 변경하지 않고 그대로 유지
        # 단순히 테이블을 생성하는 역할이며, 기존 데이터에는 영향을 주지 않음

async def get_db():
    async with AsyncSessionLocal() as session: # async with 블록을 사용하여 세션이 자동으로 닫히도록 보장
        yield session
        # 비동기 SQLAlchemy 세션을 생성하고 반환하는 역할
        # FastAPI의 Depends()를 통해 엔드포인트에서 데이터베이스 세션을 사용 가능

        # 📌 **세션(Session)**은 데이터베이스와의 연결을 관리하는 객체로,
        #  쿼리 실행, 데이터 추가, 수정, 삭제 등의 작업을 할 때 사용됨.

class UserRegister(BaseModel):
    username : str
    password : str

class UserLogin(BaseModel):
    username : str
    password : str

class Token(BaseModel):
    access_token : str
    token_type : str

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# Passlib 라이브러리를 사용해 비밀번호를 안전하게 해싱하고 검증하기위한 설정
# bcrypt 알고리즘을 사용하여 비밀번호 암호화
# deprecated 사용해 이전 해싱 방식이 자동으로 업데이트 되도록 설정

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")
# OAuth2 인증 방식 중 “Password Flow” 방식을 사용하여 액세스 토큰을 발급하고 인증을 수행
# OAuth2PasswordBearer(tokenUrl="login") → 사용자가 로그인할 때 JWT 토큰을 발급하는 URL을 지정
# FastAPI의 Depends(oauth2_scheme)를 통해 자동으로 요청 헤더에서 토큰을 추출 가능

def hash_password(password: str):
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str):
    return pwd_context.verify(plain_password, hashed_password)

SECRET_KEY = "secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def create_access_token(data: dict): # JWT 엑세스 토큰을 생성하는 함수 
    to_encode = data.copy() # data 딕셔너리를 복사하여 변경해도 원본 데이터가 영향을 받지 않도록 보호
    expire = datetime.now() + timedelta(minutes =  ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire}) 
    # to_encode 딕셔너리에 "exp"(만료 시간) 키 추가
    # JWT를 생성할 때 만료 시간이 포함되어야 자동으로 검증 가능
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) # JWT를 생성하는 핵심 코드
    return encoded_jwt

@app.post("/register")
async def register(user: UserRegister, db : AsyncSession = Depends(get_db)):
    new_user = User(username=user.username, password= hash_password(user.password))
    db.add(new_user)
    await db.commit()
    await db.refresh(new_user)

    return new_user

@app.post("/login")
async def login_user(user: UserLoginRegister, db: AsyncSession= Depends(get_db)):
    result = await db.execute(select(User).where(User.username == user.username))
    existing_user = result.scalars().first()
    if not existing_user or not verify_password(user.password, existing_user.password):
        raise HTTPException(
            status_code=404,
            detail="User not found"
        )
    access_token = create_access_token(data={"sub": existing_user.username})
    return {"access_token": access_token, "token_type": "bearer"}

async def get_current_user(token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db)):
    try:
        # 1. JWT 토큰 디코딩
        # SECRET_KEY와 알고리즘을 사용하여 토큰을 디코딩합니다.
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])

        # 2. 사용자 이름 추출
        # 토큰 페이로드에서 "sub" (subject), "exp" 필드를 가져옵니다.
        username = payload.get("sub")
        exp = payload.get("exp")

        result = await db.execute(select(User).filter(User.username == username))
        user = result.scalars().first()

        # 3. 토큰이 유효하지 않거나 sub 필드가 없으면 오류 반환
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid token")

        # 4. 사용자 검증
        # 데이터베이스(또는 가짜 DB)에서 사용자가 존재하는지 확인합니다.
        if not user:
            raise HTTPException(status_code=401, detail="User not found")

        # 5. 토큰 만료 시간 검증
        # exp시간을 확인합니다.
        if exp is None or datetime.utcnow() > datetime.utcfromtimestamp(exp):
            raise HTTPException(status_code=401, detail="Token has expired")

        # 6. 검증된 사용자 정보 반환
        return user

    except JWTError:
        # 7. 토큰 검증 실패 시 오류 반환
        raise HTTPException(status_code=401, detail="Invalid token")

@app.get("/profile")
async def get_profile(current_user: dict = Depends(get_current_user)):
    return current_user
# 현재 로그인한 사용자의 토큰 만료 시간을 가져오는 GET /token-expiry API
@app.get("/token_expiry")
async def token_expiry(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        exp = payload.get("exp")
        if exp is None:
            raise HTTPException(status_code=401, detail="Token missing expiration time")

        if datetime.utcnow() > datetime.utcfromtimestamp(exp):
            raise HTTPException(status_code=404, detail="Expired token")

        return {"expire_time": datetime.utcfromtimestamp(exp)}
    except ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Expired token")
    except JWTError:
        raise HTTPException(status_code=404, detail="Invalid token")

# 관리자만 접근 가능한 GET /admin API
@app.get("/admin/")
async def admin(current_user: dict = Depends(get_current_user)):
    if current_user.role == "admin":
        return {"msg":"welcome admin"}
    raise HTTPException(status_code=403, detail="Forbidden")

# 사용자의 프로필을 수정하는 POST /profile API
@app.put("/profile/")
async def put_profile(
        update_user: UserLogin,
        current_user: dict = Depends(get_current_user),
        db: AsyncSession = Depends(get_db)
):
    result = await db.execute(select(User).filter(User.username == current_user.username))
    existing_user = result.scalars().first()

    if not existing_user:
        raise HTTPException(status_code=404, detail="Its not your profile.")
    existing_user.username = update_user.username
    await db.commit()
    await db.refresh(existing_user)

    return {"message": "Profile updated successfully", "username": existing_user.username}

비동기 DB 작업

• 데이터베이스 요청(ex. 데이터 삽입, 조회, 수정 등)을 처리할 때 I/O 작업(ex. 네트워크 또는 파일 시스템 접근)이 발생
• 이 I/O 작업 동안 애플리케이션이 멈추지 않고 다른 작업을 처리할 수 있도록 비차단 방식을 사용

=> 성능 향상, 동시성 지원 ,자원 절약

비동기 await

• 비동기 함수 내에서 순차적으로 실행하지만
• 같은 비동기 함수가 gather로 묵여서 실행되었을때 순차적으로 진행되는걸 지정
• await 함수가 실행중이면 그 밑의 작업은 진행하지 않음, 같이 실행된 다른 비동기 함수는 동작중

비동기 DB 작업과 SQLAlchemy

• 비동기 DB 설정
  • create_async_engine
  • AsyncSession

    기본설정

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession 
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker 
from sqlalchemy import Column, Integer, String
# 비동기 SoLite 데이터베이스 URL 설정
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
# 비동기 SQLAtchemy 엔진 생성
async_engine = create_async_engine (
    DATABASE_URL,
    echo=True, # SOL 쿼리 로그 출력
)
# 비동기 세션 생성
AsyncSessionLocal = sessionmaker (
    bind=async_engine,
    class_=AsyncSession,
    expire_on_commit=False
)
# Base 클래스 생성
Base = declarative_base()

• 데이터베이스 초기화
  • 테이블 생성: 동일하게 설정
  • 엔진 초기화: async with 사용

    async def init_db():
    async with async_engine.begin() as conn:
        await conn. run_sync (Base.metadata.create_ all)
    

데이터베이스 초기화 실행

if name == "main": import asyncio asyncio. run (init_db())

### 🧑‍💻CRUD 구현
```python
from sqlalchemy import Column, Integer, String, text, select
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "sqlite+aiosqlite:///./test.db"

async_engine = create_async_engine(
    DATABASE_URL,
    echo=True, # 쿼리 로그 출력
)

# 비동기 세션 생성
AsyncSessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    expire_on_commit=False,
    class_=AsyncSession,
)

# Base클래스 생성
Base = declarative_base()

# 데이터베이스 모델 정의
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

# 데이터베이스 초기화 함수
async def init_db(): # 테이블 생성 함수
    async with async_engine.connect() as conn:
    # async with : 연결을 자동으로 열고 닫는 역할(자원 누수를 방지)
    # async with 블록이 끝나면 자동으로 conn.close()가 호출됨.
    # 즉, async with는
    # try-finally처럼 동작하여 자원을 자동으로 정리함
    # 데이터베이스와 연결하는 코드

        await conn.run_sync(Base.metadata.create_all)
        # 모델을 생성하는 함수
        # await conn : 비동기 SQLAlchemey에서 동기 코드를 실행할 떄 사용하는 방식

# 비동기 세션 생성 함수
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session
        # ✔ await은 비동기 작업이 끝날 때까지 기다리는 역할을 함.
        # ✔ yield는 함수의 실행을 멈추고, 나중에 다시 실행을 이어갈 수 있도록 하는 역할을 함.

# CRUD

# CREATE
async def create_user(name: str, email : str):
    async with AsyncSessionLocal() as session:
        new_user = User(name=name, email=email)
        session.add(new_user)
        await session.commit()
        await session.refresh(new_user)
        return new_user
# READ
async def read_user():
    async with AsyncSessionLocal() as session:
        result = await session.execute(text("SELECT * FROM users"))
        result2 = await session.execute(select(User))
        users = result.fetchall()
        users2 = result2.scalars().all()
        return users2
    # users = await session.query(User).all() # 비동기에서 사용 불가

async def read_user_by_email(email: str):
    async with AsyncSessionLocal() as session:
        result = await session.execute(
            text("SELECT * FROM users WHERE email = :email"), {"email": email}  # ✅ 바인딩 적용
        )
        result2 = await session.execute(select(User).filter(User.email == email))
        user = result.fetchone()
        user2 = result2.scalars().first()
        return user2

# PUT
async def update_user(user_id: int, name: str, email: str):
    async with AsyncSessionLocal() as session:
        result = await session.execute(
            text("SELECT * FROM users WHERE id = :user_id"),{"user_id": user_id} 
        )
        result2 = await session.execute(select(User).filter(User.id == user_id))
        user3 = await session.get(User, user_id)

        user = result.fetchone()
        user2 = result2.scalars().first()

        user2.name = name
        user2.email = email
        if not user:
            return None

        await session.commit()
        await session.refresh(user2)

        return user2

# DELETE
async def delete_user(user_id :int):
    async with AsyncSessionLocal() as session:
        # result = await session.execute(
        #     text("DELETE FROM users WHERE id = :user_id"),{"user_id": user_id} 
        # )

        user = await session.get(User, user_id)
        if not user:
            return None
        await session.delete(user)
        await session.commit()

await의 정확한 동작

• await 함수 전체가 아니라 특정 비동기 작업을 기다린다!!!
  • await를 붙이면 해당 비동기 작업이 끝날때까지 실행을 멈추고 기다린다
  • 같은 비동기 함수 안의 다른코드들은 병렬적으로 실행될 수 있음

🧑‍💻 FastAPI 기반 CRUD 구현

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy import Column, Integer, String, select
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, session
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

DATABASE_URL = "sqlite+aiosqlite:///./test.db"

async_engine = create_async_engine(
    DATABASE_URL,
    echo = True
)

AsyncSessionLocal = sessionmaker(
    bind = async_engine,
    expire_on_commit = False,
    class_=AsyncSession
)

Base = declarative_base()

app = FastAPI()

class Animal(Base):
    __tablename__ = "animals"

    id = Column(Integer, primary_key=True, index=True)
    species = Column(String, index=True)
    age = Column(Integer, index=True)
    name = Column(String, index=True)

async def init_db():
        async with async_engine.connect() as conn:
            await conn.run_sync(Base.metadata.create_all)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

# 등록된 모든 동물 목록을 반환
@app.get("/animals/")
async def get_all_animals(db: AsyncSession = Depends(get_db)):
    results = await db.execute(select(Animal))
    animals = results.scalars().all()
    return animals

# 특정 동물 종의 세부 정보 조회
@app.get("/animals/by-species")
async def get_animals_by_species(species: str=None, db: AsyncSession = Depends(get_db)):
    if species:
        result = await db.execute(select(Animal).filter(Animal.species==species))
        animals = result.scalars().all()
        if not animals :
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
        return animals
    return []


# 특정 나이 이상의 동물 조회
@app.get("/animals/by-min-age")
async def get_animals_by_min_age(min_age:int=None, db: AsyncSession = Depends(get_db)):
    if min_age:
        results = await db.execute(select(Animal).filter(Animal.age >= min_age))
        animals = results.scalars().all()
        if not animals:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
        return animals
    return []

# 특정 문자열의 이름을 가진 동물 검색
@app.get("/animals/search-by-name")
async def get_animals_by_name(name_contains: str=None, db: AsyncSession= Depends(get_db)):
    if name_contains:
        results = await db.execute(select(Animal).filter(Animal.name.like(f"%{name_contains}%")))
        animals = results.scalars().all()
        if not animals:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
        return animals
    return []

# 동물 목록을 나이순(오름차순)으로 정렬
@app.get("/animals/age_asc")
async def get_animals_by_asc(db : AsyncSession = Depends(get_db)):
    results = await db.execute(select(Animal).order_by(Animal.age))
    animals = results.scalars().all()
    return animals

# 특정 id의 동물 정보 조회
@app.get("/animals/{animal_id}")
async def get_animal_by_id(animal_id: int, db: AsyncSession = Depends(get_db)):
    # results = await db.execute(select(Animal).filter(Animal.id == animal_id))
    # animal = results.scalars().first()
    animal = await db.get(Animal, animal_id)
    if not animal:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
    return animal

• poetry add sqlalchemy

• Engine
  • 데이터베이스와의 연결을 관리하는 핵심 요소
  • 데이터베이스 URL을 통해 연결 설정

    from sqlalchemy import create_engine
    engine = create_engine("sqlite:///./test.db")
    

보통

DATABASE_URL = "sqlite:///./test.db" engine = create_engine( DATABASE_URL, echo=True )

* Session
  - 데이터베이스와의 작업(삽입, 조회, 수정, 삭제)을 관리하며,트랜잭션을 제어
  - 여러 작업을 하나의 트랜잭션으로 묶어 처리 가능

* Base
  - 테이블 정의를 위한 클래스 기반
  - 모든 테이블 클래스는 Base를 상속받아 정의
```python
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

• SQLAlchemy 기본 문법

  • 데이터베이스 생성

    Base.metadata.create_all(bind=engine)

• 함수선언시 실행방법

  @app.on_event("startup")
  async def startup():
    await init_db()

  ForeginKey, 역참조

• relationship : 역참조할 명 서로 지정해줘야함

  • 첫번쨰 인자 :

    • SQL에선 없는 타입이라 SQLAlchemy 내에서 찾아야함
    • 해당 모델 클래스명 입력

  • 두번째인자 back_populates=

    • 반대편 필드의 ForeginKey로 지정 된 것을 지정
    • 어떤 필드와 연결될건지 지정

      1:N 관계

      class User(Base):
      __tablename__ = 'users'
      

    user_id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4())) username = Column(String, nullable=False) email = Column(String, nullable=False) password = Column(String, nullable=False) role = Column(String, default = "user") is_active = Column(Boolean)

    첫번째인자: 연결할 모델 클래스명, 유저에서 찾는거니까,

    생성시에 역참조할거는 아직 DB에 없으니까 함수내에서 찾아야함 그래서 클래스명 자체를 쓰는거임 Order

    sql 에선 실재로 이런관계가 없으니 sqlalchemy 자체에서 찾아야함 그래서 클래스명으로 사용

    두번째인자: Order테이블의 user를 참조한것

    relationship()은 ForeignKey()와 다르게 SQLAlchemy ORM 내부에서 참조할 필드명을 사용해야 하기 때문에 user를 써야 함

    orders = relationship("Order", back_populates="user", cascade="all, delete")

class Order(Base): tablename = 'orders'

order_id = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String, nullable=False)
total_price = Column(DECIMAL(65, 2))
is_paid = Column(Boolean)
created_at = Column(DATETIME, default=datetime.utcnow)

# 실제 users의 DB테이블 이름을 가르킴, 이건 SQL내에서 실행되야 하는거니까

user_id = Column(ForeignKey('users.user_id'),nullable=False)
user = relationship("User", back_populates="orders")

#### N:N 관계 : secondary 에 두 테이블을 연결할 모델이 필요함
```python
order_products = Table(
    "order_products",
    Base.metadata,
    Column("order_id", ForeignKey("orders.order_id"), primary_key=True),
    Column("product_id", ForeignKey("products.product_id"), primary_key=True)
)

class Order(Base):
    __tablename__ = 'orders'

    order_id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String, nullable=False)
    total_price = Column(DECIMAL(65, 2))
    is_paid = Column(Boolean)
    created_at = Column(DATETIME, default=datetime.utcnow)

    # 실제 users의 DB테이블 이름을 가르킴, 이건 SQL내에서 실행되야 하는거니까

    user_id = Column(ForeignKey('users.user_id'),nullable=False)
    user = relationship("User", back_populates="orders")

    # Product 과 참조
    products = relationship("Product", secondary=order_products, back_populates="orders")


class Product(Base):
    __tablename__ = 'products'

    product_id = Column(String, primary_key=True)
    name = Column(String, nullable=False)
    price = Column(Float, nullable=False)
    discount = Column(Float)
    final_price = Column(Float)

    # Order와 참조
    orders = relationship("Order", secondary=order_products, back_populates="products")

• FastAPI : joinedload() = Django : select_related(), JOIN

  • JOIN을 사용하여 한 번의 쿼리로 데이터를 가져옴 (즉시 로드)
  • 1:N 관계에서 N이 많으면 중복 데이터가 많아지고 비효율적
  • User 1개에 대한 모든 주문 내역을 한 번에 가져올 때 적합
  • 관리자가 특정 사용자 주문 내역을 보고 싶을 때

    # Django
    order = Order.objects.select_related("users").get(order_id=1)
    # FastAPI
    result = await db.execute(
        select(Order).options(joinedload(Order.users)).where(Order.Order_id == Order_id)
    )

• FastAPI : selectinload() = Django : prefetch_related(), 전부 가져옴 한번에

  • IN 서브쿼리를 사용하여 여러 개의 데이터를 한 번에 로드 (즉시 로드)
  • 1:N 관계에서 N이 많을 경우 JOIN보다 효율적
  • User 목록을 가져오면서 각 사용자의 주문도 같이 로드할 때 적합
  • 여러 개의 User에 대해 Order 데이터를 한꺼번에 가져올 때 사용

    # Django
    user = User.objects.prefetch_related("orders").get(user_id=1)
    # FastAPI
    result = await db.execute(
        select(User).options(selectinload(User.orders)).where(User.user_id == user_id)
    )

• API

  @orders_router.post("/")
  async def create_order(
        order: OrderSchema,
        db: AsyncSession=Depends(get_db),
        user : dict = Depends(get_current_user)
    ):
    products = await db.execute(select(Product).where(Product.product_id.in_(order.product_id)))
    products = products.scalars().all()
  
    if not products:
        raise HTTPException(status_code=400, detail="No valid products found")
  
    new_order = Order(
        username = user.username,
        total_price = order.total_price,
        is_paid = order.is_paid
    )
  
    db.add(new_order)
    await db.commit()
    await db.refresh(new_order)
    # 주문이 없는 관계에서 바로 추가하면 주문이 없는 상태이기에 에러가 발생 그래서 따로 생성해야함
    # Many-to-Many 관계(relationship(secondary=order_products))에서 .extend()를 호출하면,
    # SQLAlchemy ORM이 내부적으로 변경을 감지하고 트랜잭션에 포함함
    # 이 트랜잭션을 DB에 반영하려면 반드시 db.commit()을 호출해야 함
    new_order.products.extend(products)
    await db.commit()
    await db.refresh(new_order)
    return new_order
  
  

@orders_router.get( "/", response_model=List[OrderSchema], response_model_exclude={ "order_id", "is_paid", "created_at" } ) async def current_user_orders(db: AsyncSession = Depends(get_db), user: dict = Depends(get_current_user)): orders = await db.execute( select(Order) .options(selectinload(Order.products)) .where(Order.username == user.username) ) orders = orders.scalars().all()

return orders

#### 🧑‍💻 총 코드
```python
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine("sqlite:///./test.db")
SessionLocal = sessionmaker (bind=engine)
session = SessionLocal()

Base = declarative_base()
Base.metadata.create_all(bind=engine)

🧑‍💻 CRUD

# 데이터 삽입
session = SessionLocal()
new_user = User(name="John")
session.add(new_user)
session.commit()

# 조회
users = session.query(User).all()
for user in users:
    print(user.name)
# 조건조회
user = session.query(User).filter(User.name =='John').first()
print(user.id, user.name, user.email)

# 수정
user = session.query(User).filter(User.name =='John').first()
if user:
    user.email = "new@new.com"
    session.commit()
# 삭제
if user:
    session.delete(user)
    session.commit()

🧑‍💻 실습

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
    DATABASE_URL,
    connect_args={"check_same_thread": False}, # 멀티 쓰레드 활성화
)

SessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine
)

Base = declarative_base()

app = FastAPI()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True,index=True)

Base.metadata.create_all(bind=engine)

class UserResponse(BaseModel):
    id : int
    name: str
    email: str

def get_db(): # 데이터베이스 연결
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/name")
def read_username(db:Session = Depends(get_db)):
    users = db.query(User.name).order_by(User.name, User.id.desc()).all()
    return [user[0] for user in users]

@app.get("/users/count")
def read_username(db:Session = Depends(get_db)):
    users = db.query(User.name).order_by(User.name, User.id.desc()).all()
    return len(users)
@app.get("/users/")
def read_users(skip: int = 0, limit: int = 10,db: Session = Depends(get_db)):
    users = db.query(User).order_by(User.name, User.id.desc()).offset(skip).limit(limit).all()
    return users
# offset(skip): skip 개수만큼 건너뛰고 데이터를 가져옴.
# limit(limit): 최대 limit 개수만큼 데이터만 가져옴.
@app.post("/users/add/")
def add_user(user : UserResponse, db : Session = Depends(get_db)):
    new_user = User(id=user.id, name=user.name, email=user.email)
    db.add(new_user)
    db.commit()
    return {
        "msg" : "success create"
    }

@app.get("/users/search")
def read_search_user(email : str= None, name: str =None, db: Session = Depends(get_db)):
    if email:
        user = db.query(User).filter(User.email == email).first()
        if user:
            return user
    if name:
        user = db.query(User).filter(User.name.like(f"%{name}%")).first()
        if user:
            return user
    raise HTTPException(status_code=404, detail="User not found")



@app.get("/users/{user_id}")
def read_user(user_id: int, db : Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.get("/users/ge/{user_id}")
def read_ge_user(user_id : int, db : Session = Depends(get_db)):
    user = db.query(User).filter(User.id >= user_id).all()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

• Dependency Injection 개념
  • 특정 동작에 필요한 의존성을 함수나 클래스 외부에서 주입하는 패턴
  • FastAPI에서는 Depends를 사용하여 필요한 의존성을 정의하고 관리할 수 있음
  • 주로 데이터베이스 세션, 인증, 설정 등의 관리에서 사용됨
  • 특정 함수를 엔드포인트에 자동으로 주입

• 연결 가능 목록

  • 데이터베이스 세션(DB connection)

    @app.get("/items/")
    async def read_items(db: AsyncSession = Depends(get_db)):  # ✅ DB 세션 자동 주입
    result = await db.execute("SELECT * FROM items")
    items = result.fetchall()
    return items

  • JWT토큰을 통한 사용자 인증

    SECRET_KEY = "secret-key"
    ALGORITHM = "HS256"
    

  oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

  async def get_current_user(token: str = Depends(oauth2_scheme)): # ✅ 토큰 자동 주입

  try:
      payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
      username = payload.get("sub")
      if not username:
          raise HTTPException(status_code=401, detail="Invalid token")
  except Exception:
      raise HTTPException(status_code=401, detail="Invalid token")
  
  return {"username": username}

  @app.get("/profile/") async def read_profile(current_user: dict = Depends(get_current_user)): # ✅ 현재 사용자 정보 자동 주입

  return {"message": f"Hello, {current_user['username']}!"}

  - 환경 변수, 설정 값
  ```python
  
  class AppSettings:
    def __init__(self):
        self.debug = True
        self.api_key = "my-secret-api-key"
  
  async def get_settings():
    return AppSettings()
  
  @app.get("/settings/")
  async def read_settings(settings: AppSettings = Depends(get_settings)):  # ✅ 설정 값 자동 주입
      return {"debug": settings.debug, "api_key": settings.api_key}

  • 공통적으로 실행할 미들웨어 기능(예: 로깅, 보안 검사)

    from fastapi import Request, Depends
    

  async def log_request(request: Request):

  print(f"📌 요청 수신: {request.method} {request.url}")
  return request

  @app.get("/secure-data/") async def secure_data(request: Request = Depends(log_request)): # ✅ 요청 로깅 자동 적용

  return {"message": "This is secure data!"}

  - OAuth2 인증 시스템(OAuth2PasswordBearer)
  ```python
  from fastapi.security import OAuth2PasswordBearer
  
  oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
  
  @app.get("/users/me/")
  async def read_users_me(token: str = Depends(oauth2_scheme)):  # ✅ 토큰 자동 주입
      return {"token": token}