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

Represents the acceptable types of a redis key

alias of str | bytes

ValueT

Represents the different python primitives that are accepted as input parameters for commands that can be used with loosely defined types. These are encoded using the configured encoding before being transmitted.

alias of str | bytes | int | float

StringT

The canonical type used for input parameters that represent “strings” that are transmitted to redis.

alias of str | bytes

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

Restricted union of container types accepted as arguments to apis that accept a variable number values for an argument (such as keys, values). This is used instead of typing.Iterable as the latter allows str to be passed in as valid values for Iterable[str] or bytes to be passed in as a valid value for Iterable[bytes] which is never the actual expectation in the scope of coredis. For example:

def length(values: Parameters[ValueT]) -> int:
    return len(list(values))

length(["1", 2, 3, 4])      # valid
length({"1", 2, 3, 4})      # valid
length(("1", 2, 3, 4))      # valid
length({"1": 2}.keys())     # valid
length({"1": 2}.values())   # valid
length(map(str, range(10))) # valid
length({"1": 2})            # invalid
length("123")               # invalid
length(b"123")              # invalid

alias of list[T_co] | Set[T_co] | tuple[T_co, …] | ValuesView[T_co] | Iterator[T_co]

Redis Response (RESP) descriptions

The follow two types describe the total representation of parsed responses from the redis serialization protocol(s) (RESP & RESP3) (See 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.

ResponsePrimitive

Mapping of primitives returned by redis

alias of str | bytes | int | float | bool | None

ResponseType

alias of str | bytes | int | float | bool | None | list[Any] | MutableSet[str | bytes | int | float | bool | None | tuple[str | bytes | int | float | bool | None, …] | frozenset[str | bytes | int | float | bool | None]] | dict[str | bytes | int | float | bool | None | tuple[str | bytes | int | float | bool | None, …] | frozenset[str | bytes | int | float | bool | None], Any] | RedisError

Response Types

In most cases the API returns native python types mapped as closely as possible to the response from redis. The responses are normalized across RESP versions 2 and 3 to maintain a consistent signature (Most notable example of this is dictionary

In certain cases these are “lightly” typed using NamedTuple or TypedDict for ease of documentation and in the case of “tuples” returned by redis - to avoid errors in indexing.

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

description: StringT

the library’s description

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

class RoleInfo(role: str, offset: int | None = None, status: str | None = None, slaves: Tuple[Dict[str, str | int], ...] | None = None, masters: Tuple[str, ...] | None = None)[source]

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, maximum_identifier: StringT, consumers: OrderedDict[StringT, int])[source]

Bases: NamedTuple

Summary response from XPENDING

pending: int

Alias for field number 0

minimum_identifier: StringT

Alias for field number 1

maximum_identifier: StringT

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: Tuple[StringT, ...]

The array composing the arguments of the command.

client_addr: Tuple[StringT, int]

Client IP address and port

client_name: str

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, ...]

matches

length: int

Length of longest match

class MonitorResult(time: datetime, db: int, client_addr: tuple[str, int] | str | None, client_type: Literal['tcp', 'unix', 'lua'], command: str, args: tuple[str, ...] | None)[source]

Bases: object

Details of issued commands received by the client when listening with the MONITOR command

time: datetime

Time command was received

db: int

db number

client_addr: tuple[str, int] | str | None

(host, port) or path if the server is listening on a unix domain socket

client_type: Literal['tcp', 'unix', 'lua']

The type of the client that send the command

command: str

The name of the command

args: tuple[str, ...] | None

Arguments passed to the command

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.

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

JsonType

Type alias for valid python types that can be represented as json

alias of str | int | float | bool | None | dict[str, Any] | list[Any]

class GraphNode(id: int, labels: set[AnyStr], properties: dict[AnyStr, ResponseType])[source]

Bases: Generic

Representation of a graph node

id: int

The node’s internal ID

labels: Set[AnyStr]

A set of labels associated with the node

properties: Dict[AnyStr, ResponseType]

Mapping of property names to values

class GraphRelation(id: int, type: AnyStr, src_node: int, destination_node: int, properties: dict[AnyStr, ResponseType])[source]

Bases: Generic

Representation of a relation between two nodes

id: int

The relationship’s internal ID

type: AnyStr

Relation type

src_node: int

Source node ID

destination_node: int

Destination node ID

properties: Dict[AnyStr, ResponseType]

Mapping of all properties the relation possesses

class GraphPath(nodes: list[GraphNode], relations: list[GraphRelation])[source]

Bases: Generic

Representation of a graph path

nodes: list[GraphNode]

The nodes in the path

relations: list[GraphRelation]

The relations in the path

property path: tuple[GraphNode | GraphRelation, ...]

The path as a tuple of nodes and relations

class GraphQueryResult(header: tuple[AnyStr, ...], result_set: tuple[ResponsePrimitive | list[ResponsePrimitive | GraphNode | GraphRelation | GraphPath], ...], stats: dict[str, ResponsePrimitive])[source]

Bases: Generic

Response from GRAPH.QUERY

header: Tuple[AnyStr, ...]

List of entries in the response header

result_set: Tuple[ResponsePrimitive | List[ResponsePrimitive | GraphNode[AnyStr] | GraphRelation[AnyStr] | GraphPath[AnyStr]], ...]

The result set from the query

stats: Dict[str, ResponsePrimitive]

Mapping of query statistics

class GraphSlowLogInfo(start_time: int, command: StringT, query: StringT, duration: float)[source]

Bases: NamedTuple

Response from GRAPH.SLOWLOG

start_time: int

The unix timestamp at which the logged command was processed.

command: StringT

The array composing the arguments of the command.

query: StringT

query name

duration: float

The amount of time needed for its execution, in microseconds.