CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-redis

Python client for Redis database and key-value store

—

Pending
Overview
Eval results
Files

core-client.mddocs/

Core Client

The Redis class provides the main interface for interacting with Redis server instances. It supports all standard Redis data types and operations with comprehensive configuration options.

import redis
import ssl
from typing import TYPE_CHECKING, Optional, Union, List, Dict, Set, Tuple, Callable, Mapping, Type

from redis import Redis, ConnectionPool
from redis.credentials import CredentialProvider
from redis.retry import Retry
from redis.backoff import ExponentialWithJitterBackoff
from redis.cache import CacheInterface, CacheConfig
from redis.event import EventDispatcher

# Type checking imports
if TYPE_CHECKING:
    import OpenSSL

Capabilities

Client Initialization

Create Redis client instances with various connection options and configurations.

class Redis:
    def __init__(
        self,
        host: str = "localhost",
        port: int = 6379,
        db: int = 0,
        password: Optional[str] = None,
        socket_timeout: Optional[float] = None,
        socket_connect_timeout: Optional[float] = None,
        socket_keepalive: Optional[bool] = None,
        socket_keepalive_options: Optional[Mapping[int, Union[int, bytes]]] = None,
        connection_pool: Optional[ConnectionPool] = None,
        unix_socket_path: Optional[str] = None,
        encoding: str = "utf-8",
        encoding_errors: str = "strict",
        decode_responses: bool = False,
        retry_on_timeout: bool = False,
        retry: Retry = Retry(
            backoff=ExponentialWithJitterBackoff(base=1, cap=10), retries=3
        ),
        retry_on_error: Optional[List[Type[Exception]]] = None,
        ssl: bool = False,
        ssl_keyfile: Optional[str] = None,
        ssl_certfile: Optional[str] = None,
        ssl_cert_reqs: Union[str, "ssl.VerifyMode"] = "required",
        ssl_ca_certs: Optional[str] = None,
        ssl_ca_path: Optional[str] = None,
        ssl_ca_data: Optional[str] = None,
        ssl_check_hostname: bool = True,
        ssl_password: Optional[str] = None,
        ssl_validate_ocsp: bool = False,
        ssl_validate_ocsp_stapled: bool = False,
        ssl_ocsp_context: Optional["OpenSSL.SSL.Context"] = None,
        ssl_ocsp_expected_cert: Optional[str] = None,
        ssl_min_version: Optional["ssl.TLSVersion"] = None,
        ssl_ciphers: Optional[str] = None,
        max_connections: Optional[int] = None,
        single_connection_client: bool = False,
        health_check_interval: int = 0,
        client_name: Optional[str] = None,
        lib_name: Optional[str] = "redis-py",
        lib_version: Optional[str] = None,
        username: Optional[str] = None,
        retry_on_timeout: bool = False,
        retry_on_error: Optional[List[Type[Exception]]] = None,
        redis_connect_func: Optional[Callable[[], None]] = None,
        credential_provider: Optional[CredentialProvider] = None,
        protocol: Optional[int] = 2,
        cache: Optional[CacheInterface] = None,
        cache_config: Optional[CacheConfig] = None,
        event_dispatcher: Optional[EventDispatcher] = None
    ): ...

    @classmethod
    def from_url(
        cls,
        url: str,
        **kwargs
    ) -> "Redis": ...

    @classmethod  
    def from_pool(
        cls,
        connection_pool: ConnectionPool
    ) -> "Redis": ...

String Operations

Redis string operations for storing and manipulating string values.

def set(
    self,
    name: KeyT,
    value: EncodableT,
    ex: Optional[ExpiryT] = None,
    px: Optional[int] = None,
    nx: bool = False,
    xx: bool = False,
    keepttl: bool = False,
    get: bool = False,
    exat: Optional[int] = None,
    pxat: Optional[int] = None
) -> Optional[bool]: ...

def get(self, name: KeyT) -> Optional[bytes]: ...

def mget(self, keys: List[KeyT], *args: KeyT) -> List[Optional[bytes]]: ...

def mset(self, mapping: Dict[KeyT, EncodableT]) -> bool: ...

def setex(self, name: KeyT, time: ExpiryT, value: EncodableT) -> bool: ...

def setnx(self, name: KeyT, value: EncodableT) -> bool: ...

def incr(self, name: KeyT, amount: int = 1) -> int: ...

def decr(self, name: KeyT, amount: int = 1) -> int: ...

def append(self, key: KeyT, value: EncodableT) -> int: ...

def substr(self, name: KeyT, start: int, end: int = -1) -> bytes: ...

def strlen(self, name: KeyT) -> int: ...

Hash Operations

Redis hash operations for storing and manipulating hash (dictionary-like) values.

def hset(
    self,
    name: KeyT,
    key: Optional[FieldT] = None,
    value: Optional[EncodableT] = None,
    mapping: Optional[Dict[FieldT, EncodableT]] = None
) -> int: ...

def hget(self, name: KeyT, key: FieldT) -> Optional[bytes]: ...

def hgetall(self, name: KeyT) -> Dict[bytes, bytes]: ...

def hmget(self, name: KeyT, keys: List[FieldT], *args: FieldT) -> List[Optional[bytes]]: ...

def hmset(self, name: KeyT, mapping: Dict[FieldT, EncodableT]) -> bool: ...

def hkeys(self, name: KeyT) -> List[bytes]: ...

def hvals(self, name: KeyT) -> List[bytes]: ...

def hlen(self, name: KeyT) -> int: ...

def hexists(self, name: KeyT, key: FieldT) -> bool: ...

def hdel(self, name: KeyT, *keys: FieldT) -> int: ...

def hincrby(self, name: KeyT, key: FieldT, amount: int = 1) -> int: ...

def hincrbyfloat(self, name: KeyT, key: FieldT, amount: float = 1.0) -> float: ...

List Operations

Redis list operations for storing and manipulating ordered collections.

def lpush(self, name: KeyT, *values: EncodableT) -> int: ...

def rpush(self, name: KeyT, *values: EncodableT) -> int: ...

def lpop(self, name: KeyT, count: Optional[int] = None) -> Union[Optional[bytes], List[bytes]]: ...

def rpop(self, name: KeyT, count: Optional[int] = None) -> Union[Optional[bytes], List[bytes]]: ...

def llen(self, name: KeyT) -> int: ...

def lrange(self, name: KeyT, start: int, end: int) -> List[bytes]: ...

def ltrim(self, name: KeyT, start: int, end: int) -> bool: ...

def lindex(self, name: KeyT, index: int) -> Optional[bytes]: ...

def lset(self, name: KeyT, index: int, value: EncodableT) -> bool: ...

def lrem(self, name: KeyT, count: int, value: EncodableT) -> int: ...

def linsert(
    self,
    name: KeyT,
    where: str,
    refvalue: EncodableT,
    value: EncodableT
) -> int: ...

def blpop(self, keys: List[KeyT], timeout: int = 0) -> Optional[Tuple[bytes, bytes]]: ...

def brpop(self, keys: List[KeyT], timeout: int = 0) -> Optional[Tuple[bytes, bytes]]: ...

Set Operations

Redis set operations for storing and manipulating unordered collections of unique elements.

def sadd(self, name: KeyT, *values: EncodableT) -> int: ...

def srem(self, name: KeyT, *values: EncodableT) -> int: ...

def smembers(self, name: KeyT) -> Set[bytes]: ...

def scard(self, name: KeyT) -> int: ...

def sismember(self, name: KeyT, value: EncodableT) -> bool: ...

def spop(self, name: KeyT, count: Optional[int] = None) -> Union[Optional[bytes], List[bytes]]: ...

def srandmember(self, name: KeyT, number: Optional[int] = None) -> Union[Optional[bytes], List[bytes]]: ...

def sunion(self, keys: List[KeyT], *args: KeyT) -> Set[bytes]: ...

def sinter(self, keys: List[KeyT], *args: KeyT) -> Set[bytes]: ...

def sdiff(self, keys: List[KeyT], *args: KeyT) -> Set[bytes]: ...

def sunionstore(self, dest: KeyT, keys: List[KeyT], *args: KeyT) -> int: ...

def sinterstore(self, dest: KeyT, keys: List[KeyT], *args: KeyT) -> int: ...

def sdiffstore(self, dest: KeyT, keys: List[KeyT], *args: KeyT) -> int: ...

Sorted Set Operations

Redis sorted set operations for storing ordered collections with scores.

def zadd(
    self,
    name: KeyT,
    mapping: Dict[EncodableT, float],
    nx: bool = False,
    xx: bool = False,
    ch: bool = False,
    incr: bool = False,
    gt: bool = False,
    lt: bool = False
) -> Union[int, float]: ...

def zrem(self, name: KeyT, *values: EncodableT) -> int: ...

def zrange(
    self,
    name: KeyT,
    start: int,
    end: int,
    desc: bool = False,
    withscores: bool = False,
    score_cast_func: Callable[[bytes], float] = float
) -> Union[List[bytes], List[Tuple[bytes, float]]]: ...

def zrevrange(
    self,
    name: KeyT,
    start: int,
    end: int,
    withscores: bool = False,
    score_cast_func: Callable[[bytes], float] = float
) -> Union[List[bytes], List[Tuple[bytes, float]]]: ...

def zcard(self, name: KeyT) -> int: ...

def zcount(self, name: KeyT, min: Union[float, str], max: Union[float, str]) -> int: ...

def zscore(self, name: KeyT, value: EncodableT) -> Optional[float]: ...

def zrank(self, name: KeyT, value: EncodableT) -> Optional[int]: ...

def zrevrank(self, name: KeyT, value: EncodableT) -> Optional[int]: ...

def zincrby(self, name: KeyT, amount: float, value: EncodableT) -> float: ...

Key Management

Redis key management operations for working with keys across all data types.

def exists(self, *names: KeyT) -> int: ...

def delete(self, *names: KeyT) -> int: ...

def expire(self, name: KeyT, time: ExpiryT) -> bool: ...

def expireat(self, name: KeyT, when: int) -> bool: ...

def ttl(self, name: KeyT) -> int: ...

def persist(self, name: KeyT) -> bool: ...

def move(self, name: KeyT, db: int) -> bool: ...

def rename(self, src: KeyT, dst: KeyT) -> bool: ...

def renamenx(self, src: KeyT, dst: KeyT) -> bool: ...

def randomkey(self) -> Optional[bytes]: ...

def type(self, name: KeyT) -> bytes: ...

def keys(self, pattern: str = "*") -> List[bytes]: ...

def scan(
    self,
    cursor: int = 0,
    match: Optional[str] = None,
    count: Optional[int] = None,
    type: Optional[str] = None
) -> Tuple[int, List[bytes]]: ...

def sort(
    self,
    name: KeyT,
    start: Optional[int] = None,
    num: Optional[int] = None,
    by: Optional[str] = None,
    get: Optional[List[str]] = None,
    desc: bool = False,
    alpha: bool = False,
    store: Optional[KeyT] = None
) -> List[bytes]: ...

Server Operations

Redis server administration and information operations.

def ping(self, **kwargs) -> Union[bytes, bool]: ...

def echo(self, value: EncodableT) -> bytes: ...

def info(self, section: Optional[str] = None) -> Dict[str, Any]: ...

def dbsize(self) -> int: ...

def flushdb(self, asynchronous: bool = False) -> bool: ...

def flushall(self, asynchronous: bool = False) -> bool: ...

def save(self) -> bool: ...

def bgsave(self, schedule: bool = True) -> bool: ...

def bgrewriteaof(self) -> bool: ...

def lastsave(self) -> int: ...

def shutdown(self, save: bool = False, nosave: bool = False) -> None: ...

def config_get(self, pattern: str = "*") -> Dict[str, str]: ...

def config_set(self, name: str, value: EncodableT) -> bool: ...

def config_resetstat(self) -> bool: ...

def config_rewrite(self) -> bool: ...

def client_list(self, type: Optional[str] = None) -> List[Dict[str, str]]: ...

def client_info(self) -> Dict[str, str]: ...

def client_setname(self, name: str) -> bool: ...

def client_getname(self) -> Optional[bytes]: ...

def client_id(self) -> int: ...

def client_kill(self, address: str) -> bool: ...

def close(self) -> None: ...

Usage Examples

Basic Key-Value Operations

import redis

# Connect to Redis
r = redis.Redis(host='localhost', port=6379, db=0)

# String operations
r.set('user:1001:name', 'John Doe')
name = r.get('user:1001:name')  # b'John Doe'

# Set with expiration
r.setex('session:abc123', 3600, 'session_data')  # Expires in 1 hour

# Increment counter
r.incr('page_views')  # Returns new count

Working with Complex Data Types

# Hash operations - user profile
r.hset('user:1001', mapping={
    'name': 'John Doe',
    'email': 'john@example.com',
    'age': '30'
})
profile = r.hgetall('user:1001')

# List operations - activity log
r.lpush('user:1001:activity', 'logged_in', 'viewed_dashboard')
recent_activity = r.lrange('user:1001:activity', 0, 4)

# Set operations - user tags
r.sadd('user:1001:tags', 'premium', 'active', 'verified')
tags = r.smembers('user:1001:tags')

# Sorted set operations - leaderboard
r.zadd('leaderboard', {'player1': 1000, 'player2': 1500, 'player3': 800})
top_players = r.zrevrange('leaderboard', 0, 2, withscores=True)

Connection from URL

import redis

# Connect using URL
r = redis.from_url('redis://localhost:6379/0')

# Connect with authentication
r = redis.from_url('redis://:password@localhost:6379/0')

# Connect with SSL
r = redis.from_url('rediss://localhost:6380/0')

Install with Tessl CLI

npx tessl i tessl/pypi-redis

docs

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