Elasticsearch를 Docker로 구축하고, **Python (FastAPI)**을 사용하여 CRUD API를 만드는 전체 과정

es-crud-project/
├── docker-compose.yml
└── main.py

version: '3.8'

services:
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.12.0
    container_name: es-container
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false  # 개발용: 보안 비활성화 (HTTPS/ID/PW 없이 접속)
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m" # 메모리 제한 (로컬 환경 부담 줄임)
    ports:
      - "9200:9200"
    networks:
      - es-net

  # (선택사항) 데이터 시각화를 위한 Kibana
  kibana:
    image: docker.elastic.co/kibana/kibana:8.12.0
    container_name: kibana-container
    environment:
      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
    ports:
      - "5601:5601"
    depends_on:
      - elasticsearch
    networks:
      - es-net

networks:
  es-net:
    driver: bridge

docker-compose up -d

pip install fastapi uvicorn elasticsearch

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from elasticsearch import Elasticsearch, NotFoundError

app = FastAPI()

# 1. Elasticsearch 연결 설정
# Docker에서 보안을 껐으므로 http://localhost:9200으로 접속합니다.
es = Elasticsearch("http://localhost:9200")

# 인덱스 이름 (RDBMS의 Table과 유사)
INDEX_NAME = "my_products"

# 데이터 모델 정의 (Request Body 검증용)
class Product(BaseModel):
    name: str
    price: int
    category: str = None

# --- CRUD API 구현 ---

# 1. CREATE (데이터 생성)
@app.post("/products/")
def create_product(product: Product):
    # document를 인덱스에 저장
    resp = es.index(index=INDEX_NAME, document=product.dict())
    return {"status": "success", "id": resp["_id"], "data": product}

# 2. READ (데이터 조회 - ID로 조회)
@app.get("/products/{item_id}")
def read_product(item_id: str):
    try:
        resp = es.get(index=INDEX_NAME, id=item_id)
        return {"id": resp["_id"], "data": resp["_source"]}
    except NotFoundError:
        raise HTTPException(status_code=404, detail="Item not found")

# 2-1. READ ALL (검색 - 모든 데이터 조회)
@app.get("/products/")
def search_products(keyword: str = None):
    if keyword:
        # 키워드가 있으면 이름에서 검색
        query = {
            "match": {
                "name": keyword
            }
        }
    else:
        # 키워드가 없으면 전체 조회 (match_all)
        query = {"match_all": {}}

    resp = es.search(index=INDEX_NAME, query=query)
    
    # 검색 결과 가공
    results = []
    for hit in resp['hits']['hits']:
        results.append({"id": hit["_id"], "data": hit["_source"]})
        
    return {"count": len(results), "results": results}

# 3. UPDATE (데이터 수정)
@app.put("/products/{item_id}")
def update_product(item_id: str, product: Product):
    try:
        # doc: 부분 업데이트할 내용
        es.update(index=INDEX_NAME, id=item_id, doc=product.dict())
        return {"status": "updated", "id": item_id, "data": product}
    except NotFoundError:
        raise HTTPException(status_code=404, detail="Item not found")

# 4. DELETE (데이터 삭제)
@app.delete("/products/{item_id}")
def delete_product(item_id: str):
    try:
        es.delete(index=INDEX_NAME, id=item_id)
        return {"status": "deleted", "id": item_id}
    except NotFoundError:
        raise HTTPException(status_code=404, detail="Item not found")

uvicorn main:app --reload