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 Union[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 Union[str, bytes, int, float]

StringT#

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

alias of Union[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 Union[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 Optional[Union[str, bytes, int, float, bool]]

ResponseType#: alias of Union[str, bytes, int, float, bool, None, list[Any], MutableSet[Union[str, bytes, int, float, bool, None, tuple[Optional[Union[str, bytes, int, float, bool]], …], frozenset[Optional[Union[str, bytes, int, float, bool]]]]], dict[Union[str, bytes, int, float, bool, None, tuple[Optional[Union[str, bytes, int, float, bool]], …], frozenset[Optional[Union[str, bytes, int, float, bool]]]], 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

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 Union[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.