Scripting¶

LUA Scripts¶

class Script(registered_client: SupportsScript | None = None, script: StringT | None = None, readonly: bool = False)[source]¶

An executable Lua script object returned by coredis.Redis.register_script(). Instances of the class are callable and take arguments in the same shape as coredis.Redis.evalsha() or coredis.Redis.eval() (i.e a list of keys and args).

Example:

client = coredis.Redis()
await client.set("test", "co")
concat = client.register_script("return redis.call('GET', KEYS[1]) + ARGV[1]")
assert await concat(['test'], ['redis']) == "coredis"

Parameters:

client¶ – The client to use for executing the lua script. If this is None the client will have to be provided when invoking the script using __call__() with the __call__.client parameter.
script¶ – The lua script that will be used by __call__()
readonly¶ – If True the script will be called with coredis.Redis.evalsha_ro() instead of coredis.Redis.evalsha()

sha: AnyStr¶: SHA of this script once it’s registered with the redis server

async __call__(keys: Parameters[KeyT] | None = None, args: Parameters[ValueT] | None = None, client: SupportsScript[AnyStr] | None = None, readonly: bool | None = None) → ResponseType[source]¶

Executes the script registered in Script.script using coredis.Redis.evalsha(). Additionally if the script was not yet registered on the instance, it will automatically do that as well and cache the sha at Script.sha

Parameters:

keys¶ – The keys this script will reference
args¶ – The arguments expected by the script
client¶ – The redis client to use instead of Script.client
readonly¶ – If True forces the script to be called with coredis.Redis.evalsha_ro()

wraps(key_spec: list[str] | None = None, param_is_key: Callable[[Parameter], bool] = lambda p: ..., client_arg: str | None = None, runtime_checks: bool = False, readonly: bool | None = None) → Callable[[Callable[[P], Awaitable[R]]], Callable[[P], Awaitable[R]]][source]¶

Decorator for wrapping a regular python function, method or classmethod signature with a Script. This allows exposing a strict signature instead of that which __call__() provides. The callable being decorated should not have an implementation as it will never be called.

The main objective of the decorator is to allow you to have strict (and type safe) signatures for wrappers for lua scripts. Internally the decorator separates keys from args before calling coredis.Redis.evalsha(). Mapping the decorated methods arguments to key providers is done either by using key_spec or param_is_key. All other paramters of the decorated function are assumed to be args consumed by the lua script.

By default the decorated method is bound to the coredis.client.Redis or coredis.client.RedisCluster instance that the Script instance was instantiated with. This may however not be the instance you want to eventually execute the method with. For such scenarios the decorated method can accept an additional parameter which has the name declared by client_arg).

The following example executes the script with the client instance used to register the script. The key parameter is detected as a key provider as it is annotated with the coredis.typing.KeyT type, and value is passed to redis as an arg:

import coredis
from coredis.typing import KeyT, ValueT
from typing import List

client = coredis.Redis()
@client.register_script("return {KEYS[1], ARGV[1]}").wraps()
async def echo_key_value(key: KeyT, value: ValueT) -> List[ValueT]: ...

k, v = await echo_key_value("co", "redis")
# (b"co", b"redis")

Alternatively, the following example builds a class method that requires the client to be passed in explicitly:

from coredis import Redis
from coredis.commands import Script

class ScriptProvider:
    @classmethod
    @Script(script="return KEYS[1]").wraps(
        key_spec=["key"],
        client_arg="client"
    )
    def echo_key(cls, client, key): ...

    @classmethod
    @Script(script="return ARGS[1]").wraps(
        client_arg="client"
    )
    def echo_arg(cls, client, value): ...

echoed = await ScriptProvider.echo_key(Redis(), "coredis")
# b"coredis"
echoed = await ScriptProvider.echo_value(Redis(), "coredis")
# b"coredis"

Parameters:

key_spec¶ – list of parameters of the decorated method that will be passed as the keys argument to __call__(). If provided this parameter takes precedence over using param_is_key to determine if a parameter is a key provider.
param_is_key¶ – a callable that accepts a single argument of type inspect.Parameter and returns True if the parameter points to a key that should be appended to the __call__.keys argument of __call__(). The default implementation marks a parameter as a key provider if it is of type coredis.typing.KeyT and is only used if key_spec is None.
client_arg¶ – The parameter of the decorator that will contain a client instance to be used to execute the script.
runtime_checks¶ – Whether to enable runtime type checking of input arguments and return values. (requires beartype). If False the function will still get runtime type checking if the environment configuration COREDIS_RUNTIME_CHECKS is set - for details see Runtime Type checking.
readonly¶ – If True forces this script to be called with coredis.Redis.evalsha_ro()

Returns:

A function that has a signature mirroring the decorated function.

Added in version 3.5.0.

Redis Functions¶

class Library(client: coredis.client.Client[AnyStr], name: StringT | None = None, code: StringT | None = None, replace: bool = False)[source]¶

Abstraction over a library of redis functions

Example:

library_code = """
#!lua name=coredis
redis.register_function('myfunc', function(k, a) return a[1] end)
"""
lib = await Library(client, "mylib", library_code)
assert "1" == await lib["myfunc"]([], [1])

When used as a base class the class variables NAME and CODE can be set on the sub class to avoid having to implement a constructor. Constructor parameters will take precedence over the class variables.

Parameters:

client¶ – The coredis client instance to use when calling the functions exposed by the library.
name¶ – The name of the library (should match the name in the Shebang in the library source).
code¶ – The lua code representing the library
replace¶ – Whether to replace the library when intializing. If False an exception will be raised if the library was already loaded in the target redis instance.

NAME: ClassVar[StringT | None] = None¶: Class variable equivalent of the Library.name argument.

CODE: ClassVar[StringT | None] = None¶: Class variable equivalent of the Library.code argument.

property functions: dict[str, Function]¶: mapping of function names to Function instances that can be directly called.

async update(new_code: StringT) → bool[source]¶: Update the code of a library with new_code

classmethod wraps(function_name: str, key_spec: list[KeyT] | None = None, param_is_key: Callable[[Parameter], bool] = lambda p: ..., runtime_checks: bool = False, readonly: bool | None = None) → Callable[[Callable[[P], Awaitable[R]]], Callable[[P], Awaitable[R]]][source]¶

Decorator for wrapping methods of subclasses of Library as entry points to the functions contained in the library. This allows exposing a strict signature instead of that which Function.__call__() provides. The callable being decorated should not have an implementation as it will never be called.

The main objective of the decorator is to allow you to represent a lua library of functions as a python class having strict (and type safe) methods as entry points. Internally the decorator separates keys from args before calling coredis.Redis.fcall().

Mapping the decorated method’s arguments to key providers is done either by using key_spec or param_is_key. All other parameters of the decorated method are assumed to be args consumed by the lua function.

The following example demonstrates most of the functionality provided by the decorator:

import coredis
from coredis.commands import Library
from coredis.typing import KeyT, ValueT
from typing import List

class MyAwesomeLibrary(Library):
    NAME = "mylib"
    CODE = """
    #!lua name=mylib

    redis.register_function('echo', function(k, a)
        return a[1]
    end)
    redis.register_function('ping', function()
        return "PONG"
    end)
    redis.register_function('get', function(k, a)
        return redis.call("GET", k[1])
    end)
    redis.register_function('hmget', function(k, a)
        local values = {}
        local fields = {}
        local response = {}
        local i = 1
        local j = 1

        while a[i] do
            fields[j] = a[i]
            i = i + 2
            j = j + 1
        end

        for idx, key in ipairs(k) do
            values = redis.call("HMGET", key, unpack(fields))
            for idx, value in ipairs(values) do
                if not response[idx] and value then
                    response[idx] = value
                end
            end
        end
        for idx, value in ipairs(fields) do
            if not response[idx] then
                response[idx] = a[idx*2]
            end
        end
        return response
    end)
    """

    @Library.wraps("echo")
    async def echo(self, value: ValueT) -> ValueT: ...

    @Library.wraps("ping")
    async def ping(self) -> str: ...

    @Library.wraps("get")
    async def get(self, key: KeyT) -> ValueT: ...

    @Library.wraps("hmmget")
    async def hmmget(self, *keys: KeyT, **fields_with_values: ValueT): ...
    """
    Return values of ``fields_with_values`` on a first come first serve
    basis from the hashes at ``keys``. Since ``fields_with_values`` is a mapping
    the keys are mapped to hash fields and the values are used
    as defaults if they are not found in any of the hashes at ``keys``
    """

client = coredis.Redis()
lib = await MyAwesomeLibrary(client, replace=True)
await client.set("hello", "world")
# True
await lib.echo("hello world")
# b"hello world"
await lib.ping()
# b"pong"
await lib.get("hello")
# b"hello"
await client.hset("k1", {"c": 3, "d": 4})
await client.hset("k2", {"a": 1, "b": 2})
await lib.hmmget("k1", "k2", a=-1, b=-2, c=-3, d=-4, e=-5)
# [b"1", b"2", b"3", b"4", b"-5"]

Parameters:

key_spec¶ – list of parameters of the decorated method that will be passed as the keys argument to __call__(). If provided this parameter takes precedence over using param_is_key to determine if a parameter is a key provider.
param_is_key¶ – a callable that accepts a single argument of type inspect.Parameter and returns True if the parameter points to a key that should be appended to the __call__.keys argument of __call__(). The default implementation marks a parameter as a key provider if it is of type coredis.typing.KeyT and is only used if key_spec is None.
runtime_checks¶ – Whether to enable runtime type checking of input arguments and return values. (requires beartype). If False the function will still get runtime type checking if the environment configuration COREDIS_RUNTIME_CHECKS is set - for details see Runtime Type checking.
readonly¶ – If True forces this function to use coredis.Redis.fcall_ro()

Returns:

A function that has a signature mirroring the decorated function.

Added in version 3.5.0.

class Function(client: coredis.client.Client[AnyStr], library_name: StringT, name: StringT, readonly: bool = False)[source]¶

Wrapper to call a redis function that has already been loaded

Parameters:

library_name¶ – Name of the library under which the function is registered
name¶ – Name of the function this instance represents
readonly¶ – If True the function will be called with coredis.Redis.fcall_ro() instead of coredis.Redis.fcall()

Example:

func = await Function(client, "mylib", "myfunc")
response = await func(keys=["a"], args=[1])

async __call__(keys: Parameters[KeyT] | None = None, args: Parameters[ValueT] | None = None, *, client: coredis.client.Client[AnyStr] | None = None, readonly: bool | None = None) → ResponseType[source]¶

Wrapper to call fcall() with the function named Function.name registered under the library at Function.library

Parameters:

keys¶ – The keys this function will reference
args¶ – The arguments expected by the function
readonly¶ – If True forces the function to use coredis.Redis.fcall_ro()