CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-redis

Python client for Redis database and key-value store

—

Pending
Overview
Eval results
Files

redis-json.mddocs/modules/

RedisJSON

RedisJSON provides native JSON data type support for Redis, allowing storage, indexing, and manipulation of JSON documents. It offers path-based operations for efficient JSON document updates without retrieving the entire document.

import redis
from typing import Any, Dict, List, Optional, Union

# Access JSON functionality through the json() method
r = redis.Redis()
json_client = r.json()

Capabilities

JSON Document Operations

Core JSON document storage and retrieval operations accessed via redis_client.json().

# JSON client obtained via r.json()
class JSON:
    def set(
        self,
        name: str,
        path: str,
        obj: Any,
        nx: Optional[bool] = False,
        xx: Optional[bool] = False,
        decode_keys: Optional[bool] = False
    ) -> Optional[str]: ...

    def get(
        self,
        name: str,
        *args: str,
        no_escape: Optional[bool] = False
    ) -> Optional[List[Any]]: ...

    def delete(self, name: str, path: str = "$") -> Optional[int]: ...

    def forget(self, name: str, path: str = "$") -> Optional[int]: ...

    def clear(self, name: str, path: str = "$") -> int: ...

    def type(self, name: str, path: Optional[str] = "$") -> List[str]: ...

    def strlen(self, name: str, path: str = "$") -> List[Optional[int]]: ...

    def toggle(self, name: str, path: str = "$") -> List[int]: ...

JSON Array Operations

Array manipulation operations for JSON arrays within documents.

def arrappend(
        self,
        name: str,
        path: Optional[str] = "$",
        *args: Any
    ) -> List[Optional[int]]: ...

    def arrindex(
        self,
        name: str,
        path: str,
        scalar: int,
        start: Optional[int] = None,
        stop: Optional[int] = None
    ) -> List[Optional[int]]: ...

    def arrinsert(
        self,
        name: str,
        path: str,
        index: int,
        *args: Any
    ) -> List[Optional[int]]: ...

    def arrlen(self, name: str, path: Optional[str] = "$") -> List[Optional[int]]: ...

    def arrpop(
        self,
        name: str,
        path: Optional[str] = "$",
        index: Optional[int] = -1
    ) -> List[Optional[str]]: ...

    def arrtrim(
        self,
        name: str,
        path: str,
        start: int,
        stop: int
    ) -> List[Optional[int]]: ...

JSON Object Operations

Object manipulation operations for JSON objects within documents.

def objkeys(self, name: str, path: Optional[str] = "$") -> List[Optional[List[str]]]: ...

    def objlen(self, name: str, path: Optional[str] = "$") -> List[Optional[int]]: ...

JSON Number Operations

Numeric operations for JSON number values.

def numincrby(
        self,
        name: str,
        path: str,
        number: int
    ) -> str: ...

    def nummultby(
        self,
        name: str,
        path: str,
        number: int
    ) -> str: ...

JSON String Operations

String manipulation operations for JSON string values.

def strappend(
        self,
        name: str,
        path: str,
        json_str: str
    ) -> List[Optional[int]]: ...

    def strlen(self, name: str, path: str = "$") -> List[Optional[int]]: ...

JSON Utility Operations

Utility operations for JSON document analysis and debugging.

def debug_memory(self, name: str, path: str = "$") -> List[int]: ...

    def resp(self, name: str, path: Optional[str] = "$") -> List[Any]: ...

Usage Examples

Basic JSON Document Operations

import redis
import json

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Store JSON document
user_data = {
    "id": 1001,
    "name": "John Doe",
    "email": "john@example.com",
    "profile": {
        "age": 30,
        "city": "New York",
        "skills": ["Python", "Redis", "JSON"]
    },
    "active": True
}

# Set JSON document
r.json().set("user:1001", "$", user_data)

# Get entire document
user = r.json().get("user:1001")
print(f"User: {user}")

# Get specific fields
name = r.json().get("user:1001", "$.name")
email = r.json().get("user:1001", "$.email")
print(f"Name: {name}, Email: {email}")

# Get nested field
city = r.json().get("user:1001", "$.profile.city")
print(f"City: {city}")

JSON Path Operations

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Complex JSON document
product_data = {
    "id": "P001",
    "name": "Gaming Laptop",
    "price": 1299.99,
    "specs": {
        "cpu": "Intel i7",
        "ram": "16GB",
        "storage": "512GB SSD"
    },
    "tags": ["gaming", "portable", "performance"],
    "reviews": [
        {"rating": 5, "comment": "Excellent!"},
        {"rating": 4, "comment": "Good performance"}
    ],
    "in_stock": True
}

r.json().set("product:P001", "$", product_data)

# Get multiple paths
result = r.json().get("product:P001", "$.name", "$.price", "$.in_stock")
print(f"Product info: {result}")

# Update specific field
r.json().set("product:P001", "$.price", 1199.99)

# Update nested field
r.json().set("product:P001", "$.specs.ram", "32GB")

# Check updated values
price = r.json().get("product:P001", "$.price")
ram = r.json().get("product:P001", "$.specs.ram")
print(f"Updated price: {price}, RAM: {ram}")

JSON Array Manipulation

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Create document with arrays
playlist_data = {
    "name": "My Favorites",
    "songs": ["Song A", "Song B", "Song C"],
    "ratings": [4.5, 3.8, 4.9],
    "metadata": {
        "created": "2024-01-01",
        "genres": ["rock", "pop"]
    }
}

r.json().set("playlist:1", "$", playlist_data)

# Append to array
r.json().arrappend("playlist:1", "$.songs", "Song D", "Song E")

# Insert into array at specific position
r.json().arrinsert("playlist:1", "$.songs", 1, "New Song")

# Get array length
length = r.json().arrlen("playlist:1", "$.songs")
print(f"Songs count: {length}")

# Find element in array
index = r.json().arrindex("playlist:1", "$.songs", "Song B")
print(f"Song B is at index: {index}")

# Pop element from array
popped = r.json().arrpop("playlist:1", "$.songs", -1)  # Pop last element
print(f"Popped song: {popped}")

# Trim array
r.json().arrtrim("playlist:1", "$.songs", 0, 2)  # Keep first 3 elements

# Check final array
songs = r.json().get("playlist:1", "$.songs")
print(f"Final songs: {songs}")

JSON Object Operations

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Create document with nested objects
config_data = {
    "database": {
        "host": "localhost",
        "port": 5432,
        "name": "myapp"
    },
    "cache": {
        "redis_host": "localhost",
        "redis_port": 6379,
        "ttl": 3600
    },
    "features": {
        "authentication": True,
        "logging": True,
        "monitoring": False
    }
}

r.json().set("config:app", "$", config_data)

# Get object keys
db_keys = r.json().objkeys("config:app", "$.database")
feature_keys = r.json().objkeys("config:app", "$.features")
print(f"Database keys: {db_keys}")
print(f"Feature keys: {feature_keys}")

# Get object lengths
db_len = r.json().objlen("config:app", "$.database")
features_len = r.json().objlen("config:app", "$.features")
print(f"Database fields: {db_len}, Feature flags: {features_len}")

# Add new fields to objects
r.json().set("config:app", "$.database.ssl", True)
r.json().set("config:app", "$.features.analytics", True)

# Check updated objects
database = r.json().get("config:app", "$.database")
features = r.json().get("config:app", "$.features")
print(f"Updated database: {database}")
print(f"Updated features: {features}")

JSON Numeric Operations

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Create document with numeric values
stats_data = {
    "user_id": 1001,
    "stats": {
        "score": 100,
        "level": 5,
        "experience": 2500.75,
        "multiplier": 1.5
    },
    "counters": {
        "games_played": 15,
        "wins": 12,
        "losses": 3
    }
}

r.json().set("user_stats:1001", "$", stats_data)

# Increment numeric values
r.json().numincrby("user_stats:1001", "$.stats.score", 25)
r.json().numincrby("user_stats:1001", "$.stats.experience", 100.25)
r.json().numincrby("user_stats:1001", "$.counters.games_played", 1)
r.json().numincrby("user_stats:1001", "$.counters.wins", 1)

# Multiply numeric values
r.json().nummultby("user_stats:1001", "$.stats.multiplier", 1.1)

# Check updated values
score = r.json().get("user_stats:1001", "$.stats.score")
experience = r.json().get("user_stats:1001", "$.stats.experience")
multiplier = r.json().get("user_stats:1001", "$.stats.multiplier")
games_played = r.json().get("user_stats:1001", "$.counters.games_played")

print(f"Score: {score}")
print(f"Experience: {experience}")
print(f"Multiplier: {multiplier}")
print(f"Games played: {games_played}")

JSON String Operations

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Create document with string values
message_data = {
    "id": "msg_001",
    "content": "Hello",
    "metadata": {
        "author": "John",
        "tags": "#important"
    }
}

r.json().set("message:001", "$", message_data)

# Append to string values
r.json().strappend("message:001", "$.content", " World!")
r.json().strappend("message:001", "$.metadata.tags", " #urgent")

# Get string lengths
content_len = r.json().strlen("message:001", "$.content")
tags_len = r.json().strlen("message:001", "$.metadata.tags")
print(f"Content length: {content_len}")
print(f"Tags length: {tags_len}")

# Check updated strings
content = r.json().get("message:001", "$.content")
tags = r.json().get("message:001", "$.metadata.tags")
print(f"Content: {content}")
print(f"Tags: {tags}")

JSON Document Management

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Create complex document
document = {
    "document_id": "doc_001",
    "title": "Project Plan",
    "sections": [
        {"name": "Introduction", "pages": 2},
        {"name": "Requirements", "pages": 5},
        {"name": "Implementation", "pages": 10}
    ],
    "metadata": {
        "created_by": "Alice",
        "version": 1.0,
        "tags": ["project", "planning"],
        "archived": False
    }
}

r.json().set("document:001", "$", document)

# Check data types
title_type = r.json().type("document:001", "$.title")
sections_type = r.json().type("document:001", "$.sections")
archived_type = r.json().type("document:001", "$.metadata.archived")

print(f"Title type: {title_type}")
print(f"Sections type: {sections_type}")
print(f"Archived type: {archived_type}")

# Toggle boolean value
r.json().toggle("document:001", "$.metadata.archived")

# Clear array (remove all elements but keep array)
r.json().clear("document:001", "$.metadata.tags")

# Delete specific path
r.json().delete("document:001", "$.metadata.version")

# Check document structure after modifications
result = r.json().get("document:001")
print(f"Modified document: {result}")

# Get memory usage (debugging)
memory_usage = r.json().debug_memory("document:001", "$")
print(f"Memory usage: {memory_usage} bytes")

Conditional JSON Operations

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Create or update with conditions
profile_data = {
    "username": "johndoe",
    "email": "john@example.com",
    "settings": {
        "theme": "dark",
        "notifications": True
    }
}

# Set only if key doesn't exist (NX - Not eXists)
created = r.json().set("profile:new_user", "$", profile_data, nx=True)
print(f"New profile created: {created}")

# Try to set again (should fail because key exists)
created_again = r.json().set("profile:new_user", "$", profile_data, nx=True)
print(f"Profile created again: {created_again}")

# Update only if key exists (XX - eXists)
updated_data = {"username": "john_doe_updated"}
updated = r.json().set("profile:new_user", "$", updated_data, xx=True)
print(f"Profile updated: {updated}")

# Try to update non-existent key (should fail)
updated_missing = r.json().set("profile:missing", "$", updated_data, xx=True)
print(f"Missing profile updated: {updated_missing}")

Bulk JSON Operations

import redis

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# Create multiple user profiles
users = [
    {"id": 1001, "name": "Alice", "department": "Engineering"},
    {"id": 1002, "name": "Bob", "department": "Sales"},
    {"id": 1003, "name": "Carol", "department": "Marketing"}
]

# Use pipeline for bulk operations
pipe = r.pipeline()

for user in users:
    pipe.json().set(f"employee:{user['id']}", "$", user)

results = pipe.execute()
print(f"Created {len(results)} employee profiles")

# Bulk retrieve specific fields
pipe = r.pipeline()
for user in users:
    pipe.json().get(f"employee:{user['id']}", "$.name", "$.department")

employee_info = pipe.execute()
for info in employee_info:
    print(f"Employee: {info}")

# Bulk update
pipe = r.pipeline()
for user in users:
    pipe.json().set(f"employee:{user['id']}", "$.last_updated", "2024-01-15")

pipe.execute()
print("Updated all employee profiles")

Install with Tessl CLI

npx tessl i tessl/pypi-redis

docs

modules

redis-bloom.md

redis-json.md

redis-search.md

redis-timeseries.md

async-support.md

cluster-support.md

connection-management.md

core-client.md

distributed-locking.md

error-handling.md

high-availability.md

index.md

pipelines-transactions.md

pubsub-messaging.md

tile.json