Typing¶

coredis.typing

Input types¶

The API uses the following type aliases to describe the unions of acceptable types for parameters to redis command wrappers.

KeyT: TypeAlias = str | bytes¶: Represents the acceptable types of a redis key

ValueT¶

Represents the different python primitives that are accepted as input parameters for commands that can be used with loosely defined types. These will eventually be serialized before being sent to redis.

Additionally any object wrapped in a Serializable will be accepted and will be serialized using an appropriate type adapter registered with the client. See Custom types for more details.

alias of str | bytes | int | float | Serializable[Any]

RedisValueT: TypeAlias = str | bytes | int | float¶: Primitive types that we can expect to be sent to redis with simple serialization. The internals of coredis pass around arguments to redis commands as this type.

StringT: TypeAlias = str | bytes¶: The canonical type used for input parameters that represent “strings” that are transmitted to redis.

MappingKeyT = ~MappingKeyT¶: Used for dictionary keys for all commands that accept Mapping as a parameter to allow any ValueT as a key

MappingStringKeyT = ~MappingStringKeyT¶: Used for dictionary keys for all commands that accept Mapping as a parameter to allow any StringT as a key

type JsonType = str | int | float | bool | dict[str, JsonType] | list[JsonType] | None¶

For methods that accept non optional variable number of keys or values, coredis does NOT use positional or keyword varargs and expects a “container” to be passed in for the argument. Common examples of such APIs are delete() and exists().

Instead of accepting Iterable, a union of select containers from the standard library are accepted via Parameters.

Parameters¶: alias of SequenceNotString[T_co] | Set[T_co] | tuple[T_co, …] | ValuesView[T_co] | Iterator[T_co]

Custom types¶

class Serializable(value: R)[source]¶

Wrapper to be used to pass arbitrary types to redis commands to be eventually serialized by coredis.typing.TypeAdapter.serialize

Wrapping a value in Serializable will pass type checking wherever a method expects a coredis.typing.ValueT - however it will still fail if there is no serializer registered through the instance of coredis.typing.TypeAdapter that is associated with the client.

For example:

class MyThing:
    ...

async with coredis.Redis() as client:
    # This will pass type checking but will fail with an :exc:`LookupError`
    # at runtime
    await client.set("fubar", coredis.typing.Serializable(MyThing()))

# however, if a serializer is registered, the above would succeed
@client.type_adapter.serializer
def _(value: MyThing) -> str:
   ... # some way to convert it to a string

class TypeAdapter[source]¶

Used by the coredis clients Redis and RedisCluster through type_adapter for adapting complex types that require custom serialization/deserialization with redis commands.

For example to use Decimal types with some common redis operations:

from decimal import Decimal
from typing import Any, Mapping, Iterable
from coredis import Redis
from coredis.typing import TypeAdapter, Serializable

adapter = TypeAdapter()

@adapter.serializer
def decimal_to_str(value: Decimal) -> str:
    return str(value)

@adapter.deserializer
def value_to_decimal(value: str|bytes) -> Decimal:
    return Decimal(value.decode("utf-8") if isinstance(value, bytes) else value)

@adapter.deserializer
def list_to_decimal_list(items: Iterable[str|bytes]) -> list[Decimal]:
    return [value_to_decimal(value) for value in items]

@adapter.deserializer
def mapping_to_decimal_mapping(mapping: Mapping[str|bytes, str|bytes]) -> dict[str|bytes, Decimal]:
    return {key: value_to_decimal(value) for key, value in mapping.items()}

async with coredis.Redis(type_adapter=adapter, decode_responses=True) as client:
    await client.set("key", Serializable(Decimal(1.5)))
    await client.lpush("list", [Serializable(Decimal(1.5))])
    await client.hset("dict", {"first": Serializable(Decimal(1.5))})
    assert Decimal(1.5) == await client.get("key").transform(Decimal)
    assert [Decimal(1.5)] == await client.lrange("list", 0, 0).transform(list[Decimal])
    assert {"first": Decimal(1.5)} == await client.hgetall("dict").transform(dict[str, Decimal])

register(type: type[R] | UnionType, serializer: Callable[[R], RedisValueT], deserializer: Callable[[Any], R], deserializable_type: type = object) → None[source]¶

Parameters:

type¶ – The type that should be serialized/deserialized
serializer¶ – a function that receives an instance of type and returns a value of type coredis.typing.RedisValueT
deserializer¶ – a function that accepts the return types from the redis commands that are expected to be used when deserializing to type.
deserializable_type¶ – the types of values deserializer should be considered for

register_serializer(serializable_type: type[R] | UnionType, serializer: Callable[[R], RedisValueT]) → None[source]¶

Parameters:

type¶ – The type that will be serialized
serializer¶ – a function that receives an instance of type and returns a value of type coredis.typing.RedisValueT

register_deserializer(deserialized_type: type[R] | UnionType, deserializer: Callable[[Any], R], deserializable_type: type | UnionType | GenericAlias = object) → None[source]¶

Register a deserializer for type and automatically register deserializers for common collection types that use this type.

Parameters:

type¶ – The type that should be deserialized
deserializer¶ – a function that accepts the return types from the redis commands that are expected to be used when deserializing to type.
deserializable_type¶ – the types of values deserializer should be considered for

serializer(func: Callable[[R], RedisValueT]) → Callable[[R], RedisValueT][source]¶

Decorator for registering a serializer

Parameters:: func¶ – A serialization function that accepts an instance of type R and returns one of the types defined by coredis.typing.RedisValueT The acceptable serializable types are inferred from the annotations in the function signature.
Raises:: ValueError – when the appropriate serializable type cannot be inferred.

deserializer(func: Callable[[Any], R]) → Callable[[Any], R][source]¶

Decorator for registering a deserializer

Parameters:: func¶ – A deserialization function that returns an instance of type R that can be used with deserialize(). The acceptable deserializable types and the expected deserialized type are inferred from the annotations in the function signature.
Raises:: ValueError – when the appropriate input/output types cannot be inferred.

serialize(value: Serializable[R]) → RedisValueT[source]¶

Serializes value into one of the types represented by RedisValueT using a serializer registered via register_serializer() or decorated by serializer().

Param:: a value wrapped in coredis.typing.Serializable

deserialize(value: Any, return_type: type[R]) → R[source]¶

Deserializes value into an instance of return_type using a deserializer registered via register_deserializer() or decorated by deserializer().

Parameters:

value¶ – the value to be deserialized (typically something returned by one of the redis commands)
return_type¶ – The type to deserialize to

Redis Response (RESP) descriptions¶

The follow types describe the total representation of parsed responses from the redis serialization protocol(s) (RESP & RESP3) (See handbook/response:redis response for more details).

In most cases these are not exposed through the client API and are only meant for internal pre-validation before the parsed response is transformed or narrowed to the returns documented in the client API at Clients.

type ResponsePrimitive = str | bytes | int | float | bool | None¶

type ResponseType = ResponsePrimitive | RedisError | list[ResponseType] | MutableSet[Hashable] | dict[Hashable, ResponseType]¶

Response Types¶

In most cases the API returns native python types mapped as closely as possible to the RESP3 type from redis:

RESP3 Type	Python type
Simple/Bulk String	`bytes` or `str` depending on the configuration of `coredis.Redis.decode_responses`
Simple/Bulk Errors	`RedisError`
Integers	`int`
Double	`float`
Arrays	`tuple`
Nulls	`NoneType`
Booleans	`bool`
Set	`set`
Map	`dict`

In certain cases the response is too complex and therefore it is “lightly” typed using NamedTuple or TypedDict for better documentation and type safety.

The types below is a complete representation of responses from coredis that are not builtin python types.

coredis.response.types

class ClientInfo¶

Bases: TypedDict

Response from CLIENT INFO

id: a unique 64-bit client ID
addr: address/port of the client
laddr: address/port of local address client connected to (bind address)
fd: file descriptor corresponding to the socket
name: the name set by the client with CLIENT SETNAME
age: total duration of the connection in seconds
idle: idle time of the connection in seconds
flags: client flags
db: current database ID
sub: number of channel subscriptions
psub: number of pattern matching subscriptions
multi: number of commands in a MULTI/EXEC context
qbuf: query buffer length (0 means no query pending)
qbuf-free: free space of the query buffer (0 means the buffer is full)
argv-mem: incomplete arguments for the next command (already extracted from query buffer)
multi-mem: memory is used up by buffered multi commands. Added in Redis 7.0
obl: output buffer length
oll: output list length (replies are queued in this list when the buffer is full)
omem: output buffer memory usage
tot-mem: total memory consumed by this client in its various buffers
events: file descriptor events
cmd: last command played
user: the authenticated username of the client
redir: client id of current client tracking redirection
resp: client RESP protocol version. Added in Redis 7.0

ScriptFlag¶

Script/Function flags See: https://redis.io/topics/lua-api#a-namescriptflagsa-script-flags

alias of Literal[‘no-writes’, ‘allow-oom’, ‘allow-stale’, ‘no-cluster’, b’no-writes’, b’allow-oom’, b’allow-stale’, b’no-cluster’]

class FunctionDefinition[source]¶

Bases: TypedDict

Function definition as returned by FUNCTION LIST

name: StringT¶: the name of the function

description: StringT¶: the description of the function

flags: set[ScriptFlag]¶: function flags

class LibraryDefinition[source]¶

Bases: TypedDict

Library definition as returned by FUNCTION LIST

name: StringT¶: the name of the library

engine: Literal['LUA']¶: the engine used by the library

functions: dict[StringT, FunctionDefinition]¶: Mapping of function names to functions in the library

library_code: StringT | None¶: The library’s source code

class ScoredMember(member: StringT, score: float)[source]¶

Bases: NamedTuple

Member of a sorted set

member: StringT¶: The sorted set member name

score: float¶: Score of the member

class GeoCoordinates(longitude: float, latitude: float)[source]¶

Bases: NamedTuple

A longitude/latitude pair identifying a location

longitude: float¶: Longitude

latitude: float¶: Latitude

class GeoSearchResult(name: StringT, distance: float | None, geohash: int | None, coordinates: GeoCoordinates | None)[source]¶

Bases: NamedTuple

Structure of a geo query

name: StringT¶: Place name

distance: float | None¶: Distance

geohash: int | None¶: GeoHash

coordinates: GeoCoordinates | None¶: Lat/Lon

class Command¶

Bases: TypedDict

Definition of a redis command See: https://redis.io/topics/key-specs

name: This is the command’s name in lowercase
arity: Arity is the number of arguments a command expects.
flags: See https://redis.io/commands/command#flags
first-key: This value identifies the position of the command’s first key name argumen
last-key: This value identifies the position of the command’s last key name argument
step: This value is the step, or increment, between the first key and last key values where the keys are.
acl-categories: This is an array of simple strings that are the ACL categories to which the command belongs
tips: Helpful information about the command
key-specification: This is an array consisting of the command’s key specifications
sub-commands: This is an array containing all of the command’s subcommands, if any

Bases: NamedTuple

Redis instance role information

role: str¶

offset: int | None¶

status: str | None¶

slaves: tuple[dict[str, str | int], ...] | None¶

masters: tuple[str, ...] | None¶

class StreamEntry(identifier: StringT, field_values: OrderedDict[StringT, StringT])[source]¶

Bases: NamedTuple

Structure representing an entry in a redis stream

identifier: StringT¶: Alias for field number 0

field_values: OrderedDict[StringT, StringT]¶: Alias for field number 1

class StreamInfo¶

Bases: TypedDict

Details of a stream See: https://redis.io/commands/xinfo-stream

class StreamPending(pending: int, minimum_identifier: StringT | None, maximum_identifier: StringT | None, consumers: OrderedDict[StringT, int])[source]¶

Bases: NamedTuple

Summary response from XPENDING

pending: int¶: Alias for field number 0

minimum_identifier: StringT | None¶: Alias for field number 1

maximum_identifier: StringT | None¶: Alias for field number 2

consumers: OrderedDict[StringT, int]¶: Alias for field number 3

class StreamPendingExt(identifier: StringT, consumer: StringT, idle: int, delivered: int)[source]¶

Bases: NamedTuple

Extended form response from XPENDING

identifier: StringT¶: Alias for field number 0

consumer: StringT¶: Alias for field number 1

idle: int¶: Alias for field number 2

delivered: int¶: Alias for field number 3

class SlowLogInfo(id, start_time, duration, command, client_addr, client_name)[source]¶

Bases: NamedTuple

id: int¶: A unique progressive identifier for every slow log entry.

start_time: int¶: The unix timestamp at which the logged command was processed.

duration: int¶: The amount of time needed for its execution, in microseconds.

command: list[StringT]¶: The array composing the arguments of the command.

client_addr: StringT¶: Client IP address:port

client_name: StringT¶: Client name

class LCSMatch(first: tuple[int, int], second: tuple[int, int], length: int | None)[source]¶

Bases: NamedTuple

An instance of an LCS match

first: tuple[int, int]¶: Start/end offset of the first string

second: tuple[int, int]¶: Start/end offset of the second string

length: int | None¶: Length of the match

class LCSResult(matches: tuple[LCSMatch, ...], length: int)[source]¶

Bases: NamedTuple

Results from LCS

matches: tuple[LCSMatch, ...]¶

length: int¶: Length of longest match

class ClusterNode[source]¶: Bases: TypedDict

class ClusterNodeDetail[source]¶: Bases: TypedDict

class PubSubMessage[source]¶

Bases: TypedDict

type: str¶

One of the following:

subscribe: Server response when a client subscribes to a channel(s)
unsubscribe: Server response when a client unsubscribes from a channel(s)
psubscribe: Server response when a client subscribes to a pattern(s)
punsubscribe: Server response when a client unsubscribes from a pattern(s)
ssubscribe: Server response when a client subscribes to a shard channel(s)
sunsubscribe: Server response when a client unsubscribes from a shard channel(s)
message: A message received from subscribing to a channel
pmessage: A message received from subscribing to a pattern

channel: StringT¶: The channel subscribed to or unsubscribed from or the channel a message was published to

pattern: StringT | None¶: The pattern that was subscribed to or unsubscribed from or to which a received message was routed to

data: int | StringT¶

If type is one of {message, pmessage} this is the actual published message
If type is one of {subscribe, psubscribe, ssubscribe, unsubscribe, punsubscribe, sunsubscribe} this will be an int corresponding to the number of channels and patterns that the connection is currently subscribed to.

class VectorData[source]¶

Bases: TypedDict

quantization: str¶: The quantization type as a string (fp32, bin or q8)

blob: bytes¶: Raw bytes representation of the vector

l2_norm: float¶: The L2 norm of the vector before normalization

quantization_range: float | None¶: If the vector is quantized as q8, the quantization range

coredis.modules.response.types

class SearchDocument(id: StringT, score: float | None, score_explanation: list[AnyStr] | None, payload: StringT | None, sortkeys: StringT | None, properties: dict[AnyStr, ResponseType])[source]¶

Bases: Generic

Search document as returned by FT.SEARCH

id: StringT¶: Document id

score: float | None¶: Search score if the withscores option was used

score_explanation: list[AnyStr] | None¶: Explanation of the score if the explainscore option was used

payload: StringT | None¶: Payload associated with the document if withpayloads was used

properties: dict[AnyStr, ResponseType]¶: Mapping of properties returned for the document

class SearchResult(total: int, documents: tuple[SearchDocument, ...])[source]¶

Bases: Generic

Search results as returned by FT.SEARCH

total: int¶: The total number of results found for the query

documents: tuple[SearchDocument, ...]¶: The documents returned by the query

class SearchAggregationResult(results: list[dict[StringT, ResponseType]], cursor: int | None)[source]¶

Bases: Generic

Search aggregations as returned by FT.AGGREGATE

results: list[dict[StringT, ResponseType]]¶: The aggregation results

cursor: int | None¶: The cursor id if with_cursor was True

class HybridResult(total_results: 'int', execution_time: 'float', warnings: 'list[AnyStr]', results: 'list[dict[AnyStr, AnyStr]]')[source]¶: Bases: Generic

class AutocompleteSuggestion(string: AnyStr, score: float | None, payload: AnyStr | None)[source]¶

Bases: Generic

Autocomplete suggestion as returned by FT.SUGGET

string: AnyStr¶: the suggestion string

score: float | None¶: the score of the suggestion if withscores was used

payload: AnyStr | None¶: the payload associated with the suggestion if withpayloads was used