Redis-OM 객체 매핑을 지원하는 redis 라이브러리

1. 정의 

   - Redis OM은 Python 애플리케이션에서 Redis를 보다 효율적으로 활용하기 위해 고수준의 추상화를 제공하는 라이브러리이다. 이를 통해 개발자는 객체 지향적인 방식으로 Redis 데이터를 모델링하고, 데이터 유효성 검사를 수행하며, 복잡한 쿼리를 간편하게 작성할 수 있다.

 

2. Redis OM의 주요 기능

   1) 객체 매핑(Object Mapping) :

      - HashModel과 JsonModel 클래스를 통해 Redis의 해시(Hash)와 JSON 데이터를 Python 객체로 매핑할 수 있다.

      - 이를 통해 Redis 데이터를 객체 지향적으로 다룰 수 있으며, Pydantic과의 통합으로 데이터 유효성 검사도 가능하다.

   2) 선언적 모델링 및 인덱싱 :

      - 데이터 모델을 선언적으로 정의하고, 필요한 필드에 대해 인덱스를 생성하여 효율적인 데이터 검색이 가능하다.

      - 예를 들어, Field(index=True)를 사용하여 특정 필드를 인덱싱할 수 있다. 

 

   3) 유연한 쿼리 API :

      - Python 표현식을 활용한 유연한 쿼리 작성이 가능하며, 복잡한 데이터 검색 작업을 간편하게 수행할 수 있다. 

 

   4) 동기 및 비동기 프로그래밍 지원 :

      - 동기식과 비동기식 프로그래밍을 모두 지원하여 다양한 애플리케이션 요구사항에 대응할 수 있다

 

   5) 데이터 유효성 검사(Data Validation) :

      - Pydantic과의 통합을 통해 입력 데이터의 유효성을 검사할 수 있다.

      - 예를 들어, 이메일 형식의 필드나 선택적(Optional) 필드를 정의하여 데이터의 무결성을 보장할 수 있다.

 

   6) 자동 키 생성 및 관리(Auto Key Generation and Management) :

      - Redis OM은 ULID를 사용하여 고유한 키를 자동으로 생성하고 관리한다.

      - 이를 통해 개발자는 키 관리에 신경 쓰지 않고 데이터 저장에 집중할 수 있다.

from redis_om import get_redis_connection, HashModel, Field
from pydantic import EmailStr
from typing import Optional

# Redis 연결 설정
redis = get_redis_connection(
    host="localhost",
    port=6379,
    decode_responses=True
)

# 데이터 모델 정의
class Student(HashModel):
    name: str
    age: int
    email: EmailStr
    about: Optional[str] = None

    class Meta:
        database = redis

# 데이터 저장
student = Student(
    name="Alice",
    age=21,
    email="alice@example.com",
    about="Computer Science major"
)
student.save()

# 데이터 조회
retrieved_student = Student.get(student.pk)
print(retrieved_student)

# 데이터 삭제
retrieved_student.delete()

 

3. Redis OM의 기능 구현 예시 

   1) 객체 매핑 및 데이터 저장 :

      - HashModel을 사용하여 Redis의 해시(Hash) 데이터 구조에 데이터를 저장한다.

from redis_om import get_redis_connection, HashModel

# Redis 연결 설정
redis = get_redis_connection(
    host="localhost",
    port=6379,
    decode_responses=True
)

# 데이터 모델 정의
class Product(HashModel):
    name: str
    price: float
    quantity: int

    class Meta:
        database = redis

# 데이터 저장
product = Product(
    name="Laptop",
    price=999.99,
    quantity=10
)
product.save()

 

   2) 데이터 유효성 검사 :

      - Pydantic의 기능을 활용하여 데이터의 유효성을 검사한다.

from redis_om import get_redis_connection, HashModel
from pydantic import EmailStr

# Redis 연결 설정
redis = get_redis_connection(
    host="localhost",
    port=6379,
    decode_responses=True
)

# 데이터 모델 정의
class User(HashModel):
    username: str
    email: EmailStr

    class Meta:
        database = redis

# 유효한 데이터 저장
user = User(
    username="john_doe",
    email="john.doe@example.com"
)
user.save()

# 유효하지 않은 이메일 형식의 데이터 저장 시도
try:
    invalid_user = User(
        username="jane_doe",
        email="invalid_email"
    )
    invalid_user.save()
except ValueError as e:
    print(f"Error: {e}")

 

   3) 선언적 인덱싱 및 쿼리 :

      - 필드에 인덱스를 설정하고, 이를 활용하여 데이터를 효율적으로 검색한다.

from redis_om import get_redis_connection, HashModel, Field

# Redis 연결 설정
redis = get_redis_connection(
    host="localhost",
    port=6379,
    decode_responses=True
)

# 데이터 모델 정의
class Employee(HashModel):
    name: str
    department: str = Field(index=True)
    salary: float

    class Meta:
        database = redis

# 데이터 저장
emp1 = Employee(name="Alice", department="HR", salary=50000)
emp2 = Employee(name="Bob", department="Engineering", salary=70000)
emp3 = Employee(name="Charlie", department="HR", salary=55000)

emp1.save()
emp2.save()
emp3.save()

# 쿼리: HR 부서의 직원 조회
hr_employees = Employee.find(Employee.department == "HR").all()
for emp in hr_employees:
    print(emp)

 

   4) 동기 및 비동기 프로그래밍 지원 :

      - 동기식과 비동기식 방식으로 데이터를 저장하고 조회할 수 있다.

import asyncio
from redis_om import get_redis_connection, HashModel

# Redis 연결 설정
redis = get_redis_connection(
    host="localhost",
    port=6379,
    decode_responses=True
)

# 데이터 모델 정의
class Task(HashModel):
    title: str
    is_completed: bool

    class Meta:
        database = redis

# 비동기 함수 정의
async def main():
    # 데이터 저장
    task = Task(title="Complete assignment", is_completed=False)
    await task.save()

    # 데이터 조회
    retrieved_task = await Task.get(task

 

- reference : 

https://redis.io/docs/latest/integrate/redisom-for-python/

RedisOM for Python