Scripting

LUA Scripts

class Script(registered_client: coredis.client.Client[AnyStr] | None = None, script: StringT | None = None, readonly: bool = False)

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()
async with client:
    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

__call__(keys: Parameters[KeyT] | None = None, args: Parameters[ValueT] | None = None, client: coredis.client.Client[AnyStr] | None = None, readonly: bool | None = None, callback: Callable[..., CommandResponseT] = NoopCallback()) → CommandRequest[CommandResponseT]

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()

• callback – a custom callback to call on the raw response from redis before returning it.

wraps(*, client_arg: str | None = None, runtime_checks: bool = False, readonly: bool = False) → Callable[[Callable[[P], Awaitable[R]]], Callable[[P], Awaitable[R]]]

wraps(*, callback: Callable[[...], CommandResponseT], client_arg: str | None = None, runtime_checks: bool = False, readonly: bool = False) → Callable[[Callable[[P], Awaitable[Any]]], Callable[[P], Awaitable[CommandResponseT]]]

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 method’s arguments to key providers is done by type annotations: all parameters annotated as KeyT will be passed as keys, and the rest will be passed as arguments.

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 coredis.commands import Command Request

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

async with client:
    res = 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
from coredis.typing import StringT

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

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

async with Redis() as client:
    echoed = await ScriptProvider.echo_key(client, "coredis")
    # b"coredis"
    echoed = await ScriptProvider.echo_value(client, "coredis")
    # b"coredis"

You can also pass a custom callback to execute on the return type, which will be inferred as the return type rather than the annotation:

class MyCallback(ResponseCallback[Any, Any, int]):
    def transform(self, response: ResponseType) -> int:
        return sum([ord(c) for c in str(response)])

client = coredis.Redis(decode_responses=True)
async with client:
    script = client.register_script("return {KEYS[1], ARGV[1]}")

    # we use Any since return type will come from callback
    @script.wraps(callback=MyCallback())
    def echo_key_value(key: KeyT, value: ValueT) -> CommandRequest[int]: ...

    res = await echo_key_value("co", "redis")
    reveal_type(res)  # int
    # 1161

Parameters:

• callback – a custom callback to execute on the returned value. When provided, the callback’s type will be inferred as the return type instead of the type from the stub.

• 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)

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: MutableMapping[str, Function]

mapping of function names to Function instances that can be directly called.

async update(new_code: StringT) → bool

Update the code of a library with new_code

wraps(*, function_name: str | None = None, runtime_checks: bool = False, readonly: bool = False, verify_existence: bool = True) → Callable[[Callable[[P], CommandRequest[R]]], Callable[[P], CommandRequest[R]]]

wraps(*, function_name: str | None = None, callback: Callable[[...], CommandResponseT], runtime_checks: bool = False, readonly: bool = False, verify_existence: bool = True) → Callable[[Callable[[P], CommandRequest[Any]]], Callable[[P], CommandRequest[CommandResponseT]]]

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 name of the function decorated must match the foreign (Lua) function’s name.

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 by type annotations: all parameters annotated as KeyT will be passed as keys, and the rest will be passed as arguments.

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

from typing import AnyStr

import coredis
from coredis.commands import CommandRequest, Library
from coredis.commands.function import wraps
from coredis.typing import KeyT, ValueT

class MyAwesomeLibrary(Library[AnyStr]):
    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 {
      function_name = 'get',
      callback = function(k, a)
        return redis.call("GET", k[1])
      end,
      flags = { 'no-writes' }  -- mark as read-only
    }
    redis.register_function('hmmget', 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)
    """

    @wraps()
    def echo(self, value: ValueT) -> CommandRequest[ValueT]: ... # type: ignore[empty-body]

    @wraps()
    def ping(self) -> CommandRequest[bytes]: ... # type: ignore[empty-body]

    @wraps(readonly=True)
    def get(self, key: KeyT) -> CommandRequest[ValueT]: ...  # type: ignore[empty-body]

    @wraps()
    def hmmget(self, *keys: KeyT, **fields_with_values: int) -> CommandRequest[list[ValueT]]: ... # type: ignore[empty-body]

async def run() -> None:
    client = coredis.Redis()
    async with client:
        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"world"

        async with client.pipeline(transaction=False) as pipe:
            pipe.hset("k1", {"c": 3, "d": 4})
            pipe.hset("k2", {"a": 1, "b": 2})
            lib = await MyAwesomeLibrary(pipe)
            res = lib.hmmget("k1", "k2", a=-1, b=-2, c=-3, d=-4, e=-5)
        print(await res)
        # [b"1", b"2", b"3", b"4", b"-5"]

Parameters:

• function_name – Optional name of the registered library function to map this entrypoint to. If not provided, the name will be inferred from the method that the decorator wraps

• callback – a custom callback to execute on the returned value. When provided, the callback’s type will be inferred as the return type instead of the type from the stub.

• 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()

• verify_existence – If False, skip check to see if the function has already been registered and call FCALL directly. Use this when you want to save a round trip and already know the function is registered.

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)

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])

__call__(keys: Parameters[KeyT] | None = None, args: Parameters[ValueT] | None = None, callback: Callable[..., CommandResponseT] = NoopCallback(), *, client: coredis.client.Client[AnyStr] | None = None, readonly: bool | None = None) → CommandRequest[CommandResponseT]

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

• callback – a custom callback to call on the raw response from redis before returning it.

• readonly – If True forces the function to use coredis.Redis.fcall_ro()