tessl/pypi-aioredis

asyncio (PEP 3156) Redis support

Overall
score

98%

Overview

Eval results

Files

Basic Operations

Name: tessl/pypi-aioredis
Author: tessl

Core Redis operations for strings, numeric values, and key management. These fundamental commands form the foundation of most Redis applications and provide essential data storage and retrieval functionality.

Capabilities

String Operations

Basic string operations for storing and retrieving text and binary data with optional expiration and conditional operations.

async def get(name: str) -> Optional[str]:
    """
    Get the string value of a key.
    
    Args:
        name: The key name
        
    Returns:
        The key value or None if key doesn't exist
    """

async def set(
    name: str, 
    value: Any, 
    ex: Optional[int] = None, 
    px: Optional[int] = None, 
    nx: bool = False, 
    xx: bool = False, 
    keepttl: bool = False, 
    get: bool = False
) -> Union[bool, Optional[str]]:
    """
    Set the string value of a key with optional parameters.
    
    Args:
        name: The key name
        value: The value to set
        ex: Expiration in seconds
        px: Expiration in milliseconds
        nx: Only set if key doesn't exist
        xx: Only set if key exists
        keepttl: Keep existing TTL
        get: Return old value
        
    Returns:
        True if successful, or old value if get=True
    """

async def mget(keys: List[str]) -> List[Optional[str]]:
    """
    Get multiple keys in a single operation.
    
    Args:
        keys: List of key names
        
    Returns:
        List of values in same order as keys
    """

async def mset(mapping: Dict[str, Any]) -> bool:
    """
    Set multiple keys in a single operation.
    
    Args:
        mapping: Dictionary of key-value pairs
        
    Returns:
        True if successful
    """

async def append(key: str, value: str) -> int:
    """
    Append value to existing string.
    
    Args:
        key: The key name
        value: Value to append
        
    Returns:
        New string length
    """

async def strlen(name: str) -> int:
    """
    Get the length of a string value.
    
    Args:
        name: The key name
        
    Returns:
        String length or 0 if key doesn't exist
    """

async def getrange(key: str, start: int, end: int) -> str:
    """
    Get substring of string stored at key.
    
    Args:
        key: The key name
        start: Start offset
        end: End offset (-1 for end of string)
        
    Returns:
        Substring
    """

async def setrange(name: str, offset: int, value: str) -> int:
    """
    Overwrite part of string at given offset.
    
    Args:
        name: The key name
        offset: Byte offset
        value: Value to write
        
    Returns:
        New string length
    """

Expiration Operations

Commands for setting and managing key expiration with various time formats and precision levels.

async def setex(name: str, time: int, value: Any) -> bool:
    """
    Set key with expiration time in seconds.
    
    Args:
        name: The key name
        time: Expiration in seconds
        value: The value to set
        
    Returns:
        True if successful
    """

async def psetex(name: str, time_ms: int, value: Any) -> bool:
    """
    Set key with expiration time in milliseconds.
    
    Args:
        name: The key name
        time_ms: Expiration in milliseconds
        value: The value to set
        
    Returns:
        True if successful
    """

async def setnx(name: str, value: Any) -> bool:
    """
    Set key only if it doesn't exist.
    
    Args:
        name: The key name
        value: The value to set
        
    Returns:
        True if key was set, False if key already exists
    """

async def msetnx(mapping: Dict[str, Any]) -> bool:
    """
    Set multiple keys only if none exist.
    
    Args:
        mapping: Dictionary of key-value pairs
        
    Returns:
        True if all keys were set, False if any key exists
    """

Numeric Operations

Atomic increment and decrement operations for numeric values with integer and floating-point support.

async def incr(name: str, amount: int = 1) -> int:
    """
    Increment key by amount.
    
    Args:
        name: The key name
        amount: Amount to increment by
        
    Returns:
        New value after increment
    """

async def incrby(name: str, amount: int = 1) -> int:
    """
    Increment key by integer amount.
    
    Args:
        name: The key name
        amount: Integer amount to increment
        
    Returns:
        New value after increment
    """

async def incrbyfloat(name: str, amount: float = 1.0) -> float:
    """
    Increment key by float amount.
    
    Args:
        name: The key name
        amount: Float amount to increment
        
    Returns:
        New value after increment
    """

async def decr(name: str, amount: int = 1) -> int:
    """
    Decrement key by amount.
    
    Args:
        name: The key name
        amount: Amount to decrement by
        
    Returns:
        New value after decrement
    """

async def decrby(name: str, amount: int = 1) -> int:
    """
    Decrement key by integer amount.
    
    Args:
        name: The key name
        amount: Integer amount to decrement
        
    Returns:
        New value after decrement
    """

Key Management

Essential key management operations for checking existence, deleting keys, setting expiration, and key introspection.

async def exists(*names: str) -> int:
    """
    Check how many keys exist.
    
    Args:
        names: Key names to check
        
    Returns:
        Number of existing keys
    """

async def delete(*names: str) -> int:
    """
    Delete one or more keys.
    
    Args:
        names: Key names to delete
        
    Returns:
        Number of keys deleted
    """

async def unlink(*names: str) -> int:
    """
    Asynchronously delete keys.
    
    Args:
        names: Key names to unlink
        
    Returns:
        Number of keys unlinked
    """

async def expire(name: str, time: int) -> bool:
    """
    Set key expiration in seconds.
    
    Args:
        name: The key name
        time: Expiration time in seconds
        
    Returns:
        True if expiration was set
    """

async def expireat(name: str, when: int) -> bool:
    """
    Set key expiration at timestamp.
    
    Args:
        name: The key name
        when: Unix timestamp
        
    Returns:
        True if expiration was set
    """

async def pexpire(name: str, time: int) -> bool:
    """
    Set key expiration in milliseconds.
    
    Args:
        name: The key name
        time: Expiration time in milliseconds
        
    Returns:
        True if expiration was set
    """

async def pexpireat(name: str, when: int) -> bool:
    """
    Set key expiration at timestamp in milliseconds.
    
    Args:
        name: The key name
        when: Unix timestamp in milliseconds
        
    Returns:
        True if expiration was set
    """

async def ttl(name: str) -> int:
    """
    Get key time to live in seconds.
    
    Args:
        name: The key name
        
    Returns:
        TTL in seconds, -1 if no expiration, -2 if key doesn't exist
    """

async def pttl(name: str) -> int:
    """
    Get key time to live in milliseconds.
    
    Args:
        name: The key name
        
    Returns:
        TTL in milliseconds, -1 if no expiration, -2 if key doesn't exist
    """

async def persist(name: str) -> bool:
    """
    Remove key expiration.
    
    Args:
        name: The key name
        
    Returns:
        True if expiration was removed
    """

async def type(name: str) -> str:
    """
    Get key data type.
    
    Args:
        name: The key name
        
    Returns:
        Key type string ('string', 'list', 'set', 'zset', 'hash', 'stream')
    """

Key Pattern Operations

Operations for finding and working with keys using patterns, random key selection, and key renaming.

async def keys(pattern: str = "*") -> List[str]:
    """
    Find keys matching pattern.
    
    Args:
        pattern: Glob-style pattern (* and ? wildcards)
        
    Returns:
        List of matching key names
    """

async def scan(
    cursor: int = 0, 
    match: Optional[str] = None, 
    count: Optional[int] = None, 
    _type: Optional[str] = None
) -> Tuple[int, List[str]]:
    """
    Incrementally scan keys.
    
    Args:
        cursor: Scan cursor position
        match: Pattern to match
        count: Hint for keys per iteration
        _type: Filter by key type
        
    Returns:
        Tuple of (next_cursor, keys)
    """

async def randomkey() -> Optional[str]:
    """
    Get random key name.
    
    Returns:
        Random key name or None if database is empty
    """

async def rename(src: str, dst: str) -> bool:
    """
    Rename key.
    
    Args:
        src: Source key name
        dst: Destination key name
        
    Returns:
        True if successful
    """

async def renamenx(src: str, dst: str) -> bool:
    """
    Rename key only if destination doesn't exist.
    
    Args:
        src: Source key name
        dst: Destination key name
        
    Returns:
        True if key was renamed, False if destination exists
    """

Bit Operations

Bitwise operations on string values for efficient bit-level data manipulation and storage.

async def getbit(name: str, offset: int) -> int:
    """
    Get bit value at offset.
    
    Args:
        name: The key name
        offset: Bit offset
        
    Returns:
        Bit value (0 or 1)
    """

async def setbit(name: str, offset: int, value: int) -> int:
    """
    Set bit value at offset.
    
    Args:
        name: The key name
        offset: Bit offset
        value: Bit value (0 or 1)
        
    Returns:
        Previous bit value
    """

async def bitcount(name: str, start: Optional[int] = None, end: Optional[int] = None) -> int:
    """
    Count set bits in string.
    
    Args:
        name: The key name
        start: Start byte offset
        end: End byte offset
        
    Returns:
        Number of set bits
    """

async def bitpos(
    key: str, 
    bit: int, 
    start: Optional[int] = None, 
    end: Optional[int] = None
) -> int:
    """
    Find first bit position.
    
    Args:
        key: The key name
        bit: Bit value to find (0 or 1)
        start: Start byte offset
        end: End byte offset
        
    Returns:
        First bit position or -1 if not found
    """

async def bitop(operation: str, dest: str, *keys: str) -> int:
    """
    Perform bitwise operation between strings.
    
    Args:
        operation: Bitwise operation ('AND', 'OR', 'XOR', 'NOT')
        dest: Destination key
        keys: Source keys
        
    Returns:
        Length of destination key
    """

Usage Examples

Basic String Operations

import aioredis

async def string_examples():
    redis = aioredis.Redis(decode_responses=True)
    
    # Basic set/get
    await redis.set('user:1:name', 'John Doe')
    name = await redis.get('user:1:name')
    print(name)  # "John Doe"
    
    # Set with expiration
    await redis.setex('session:abc123', 3600, 'session_data')
    
    # Conditional set
    created = await redis.setnx('counter', 0)  # Only if doesn't exist
    if created:
        print("Counter initialized")
    
    # Multiple operations
    await redis.mset({'key1': 'value1', 'key2': 'value2', 'key3': 'value3'})
    values = await redis.mget(['key1', 'key2', 'key3'])
    print(values)  # ['value1', 'value2', 'value3']

### Numeric Operations

```python
async def numeric_examples():
    redis = aioredis.Redis(decode_responses=True)
    
    # Initialize counter
    await redis.set('page_views', 0)
    
    # Increment operations
    await redis.incr('page_views')  # 1
    await redis.incrby('page_views', 5)  # 6
    await redis.incrbyfloat('average_score', 2.5)  # 2.5
    
    # Decrement operations
    await redis.decr('page_views')  # 5
    await redis.decrby('page_views', 2)  # 3
    
    current = await redis.get('page_views')
    print(f"Current page views: {current}")

Key Management

async def key_management_examples():
    redis = aioredis.Redis(decode_responses=True)
    
    # Check key existence
    exists = await redis.exists('user:1', 'user:2', 'user:3')
    print(f"{exists} keys exist")
    
    # Set expiration
    await redis.expire('temp_key', 300)  # 5 minutes
    remaining = await redis.ttl('temp_key')
    print(f"TTL: {remaining} seconds")
    
    # Find keys by pattern
    user_keys = await redis.keys('user:*')
    print(f"Found {len(user_keys)} user keys")
    
    # Scan for better performance with large datasets
    cursor = 0
    all_keys = []
    while True:
        cursor, keys = await redis.scan(cursor, match='session:*', count=100)
        all_keys.extend(keys)
        if cursor == 0:
            break
    
    # Key type inspection
    key_type = await redis.type('user:1:name')
    print(f"Key type: {key_type}")

Install with Tessl CLI

npx tessl i tessl/pypi-aioredis

tessl/pypi-aioredis