KVDBSession

Sessions are the underlying object that contains the state of a single KVDB connection to a server. This session contains both asynchronous and synchronous methods for interacting with the server, as well as methods for managing the session itself.

The underlying connection pools are managed by the KVDBClient class, which is a singleton that manages all sessions. This allows you to create multiple sessions with different configurations, but still share the same underlying connection pool, unless there is any configuration differences that would require a new connection pool.

API Reference

Bases: abc.ABC

The KVDB Session protocol

Initializes the KVDB Session

Source code in kvdb/components/session.py

def __init__(
    self,
    name: str,
    url: Union[str, KVDBUrl],
    *,
    pool: SessionPools,
    db_id: Optional[int] = None,
    serializer: Optional[Union['SerializerT', str]] = None,
    encoder: Optional['Encoder'] = None,
    **kwargs: Any,
) -> None:
    """
    Initializes the KVDB Session
    """
    if isinstance(url, str): url = KVDBUrl(url)
    self.name = name
    self.url = url
    self.pool = pool
    if db_id is not None and db_id != self.url.db_id:
        self.url = self.url.with_db_id(db_id)
    self.settings = settings
    self.logger = settings.logger
    self.autologger = settings.autologger
    self.init_serializer(serializer, **kwargs)
    self.init_encoder(encoder, **kwargs)
    self.init_cache_config(**kwargs)
    self.init_state(**kwargs)
    self._version: Optional[str] = None
    self._persistence: Optional['PersistentDict'] = None
    self._persistence_ctx: Dict[str, 'PersistentDict'] = {}
    self._kwargs = kwargs

aclient `property`

aclient: kvdb.components.client.AsyncKVDB

[Async] The KVDB client

client `property`

client: kvdb.components.client.KVDB

[Sync] The KVDB client

db_id `property`

db_id: typing.Optional[int]

Returns the database ID

persistence `property`

persistence: lazyops.libs.persistence.PersistentDict

Returns the PersistentDict instance that utilizes the session's client

session_serialization_enabled `property`

session_serialization_enabled: bool

Returns whether session serialization is enabled which is determined by the pool's encoder serialization to be enabled

version `property`

version: str

Returns the version

aenter `async`

__aenter__()

Enter the runtime context related to this object.

Source code in kvdb/components/session.py

async def __aenter__(self):
    """
    Enter the runtime context related to this object.
    """
    return self

aexit `async`

__aexit__(exc_type, exc_val, exc_tb)

Close the session

Source code in kvdb/components/session.py

async def __aexit__(self, exc_type, exc_val, exc_tb):
    """
    Close the session
    """
    await self.aclose()

contains

__contains__(key: kvdb.types.generic.KeyT) -> bool

[Dict] Returns whether the key exists

Source code in kvdb/components/session.py

def __contains__(self, key: KeyT) -> bool:
    """
    [Dict] Returns whether the key exists
    """
    return self.persistence.contains(key)

delitem

__delitem__(key: kvdb.types.generic.KeyT) -> None

[Dict] Deletes the key

Source code in kvdb/components/session.py

def __delitem__(self, key: KeyT) -> None:
    """
    [Dict] Deletes the key
    """
    if settings.is_in_async_loop():
        return ThreadPooler.create_background_task(self.adelitem, key)
    return self.delitem(key)

enter

__enter__()

Enter the runtime context related to this object.

Source code in kvdb/components/session.py

def __enter__(self):
    """
    Enter the runtime context related to this object.
    """
    return self

exit

__exit__(
    exc_type: (
        kvdb.components.session.KVDBSession.type[
            BaseException
        ]
        | None
    ),
    exc_val: BaseException | None,
    exc_tb: types.TracebackType | None,
)

On exit, close the session

Source code in kvdb/components/session.py

def __exit__(self, exc_type, exc_val, exc_tb):
    """
    On exit, close the session
    """
    self.close()

getitem

__getitem__(
    key: kvdb.types.generic.KeyT,
) -> kvdb.components.session.ReturnT

[Dict] Returns the value for the given key

Source code in kvdb/components/session.py

def __getitem__(self, key: KeyT) -> ResponseT:
    """
    [Dict] Returns the value for the given key
    """
    return self.getitem(key)

setitem

__setitem__(
    key: kvdb.types.generic.KeyT, value: typing.Any
) -> None

[Dict] Sets the value for the given key

Source code in kvdb/components/session.py

def __setitem__(self, key: KeyT, value: Any) -> None:
    """
    [Dict] Sets the value for the given key
    """
    if settings.is_in_async_loop():
        return ThreadPooler.create_background_task(
            self.asetitem, key, value,
        )
    return self.setitem(key, value)

aacl_cat `async`

aacl_cat(
    category: typing.Union[str, None] = None, **kwargs
) -> redis.commands.core.ResponseT

Returns a list of categories or commands within a category.

If category is not supplied, returns a list of all categories. If category is supplied, returns a list of all commands within that category.

For more information see https://redis.io/commands/acl-cat

Source code in kvdb/components/session.py

if self._persistence is None:
    from .persistence import KVDBStatefulBackend
    persistence_config = self.settings.persistence.model_dump(exclude_none = True)
    persistence_kwargs = self.settings.persistence.extract_kwargs(_prefix = 'persistence_', _exclude_none = True, **self._kwargs)
    if persistence_kwargs: persistence_config.update(persistence_kwargs)
    if 'name' not in persistence_config: persistence_config['name'] = self.name
    self._persistence = KVDBStatefulBackend.as_persistent_dict(
        session = self,
        **persistence_config,
    )

aacl_deluser `async`

aacl_deluser(
    *username: str, **kwargs
) -> redis.commands.core.ResponseT

Delete the ACL for the specified usernames

For more information see https://redis.io/commands/acl-deluser

Source code in kvdb/components/session.py

    return self.client.execute_command(*args, **options)

async def aexecute_command(self, *args: Any, **options: Any) -> Any:
    """
    Execute a command and return a parsed response
    """

aacl_dryrun `async`

aacl_dryrun(username, *args, **kwargs)

Simulate the execution of a given command by a given username.

For more information see https://redis.io/commands/acl-dryrun

Source code in kvdb/components/session.py

    return self._persistence

def execute_command(self, *args: Any, **options: Any) -> Any:
    """
    Execute a command and return a parsed response
    """

aacl_genpass `async`

aacl_genpass(
    bits: typing.Union[int, None] = None, **kwargs
) -> redis.commands.core.ResponseT

Generate a random password value. If bits is supplied then use this number of bits, rounded to the next multiple of 4. See: https://redis.io/commands/acl-genpass

Source code in kvdb/components/session.py

    return await self.aclient.execute_command(*args, **options)

"""
Component Methods
"""

aacl_getuser `async`

aacl_getuser(
    username: str, **kwargs
) -> redis.commands.core.ResponseT

Get the ACL details for the specified username.

If username does not exist, return None

For more information see https://redis.io/commands/acl-getuser

Source code in kvdb/components/session.py

def create_persistence(
    self,
    name: Optional[str] = None,
    base_key: Optional[str] = None,
    **kwargs,
) -> 'PersistentDict':
    """
    Create a new persistence instance

aacl_help `async`

aacl_help(**kwargs) -> redis.commands.core.ResponseT

The ACL HELP command returns helpful text describing the different subcommands.

For more information see https://redis.io/commands/acl-help

Source code in kvdb/components/session.py

"""
# name = name or self.name
# if name in self._persistence_ctx:
#     return self._persistence_ctx[name]

from .persistence import KVDBStatefulBackend

aacl_list `async`

aacl_list(**kwargs) -> redis.commands.core.ResponseT

Return a list of all ACLs on the server

For more information see https://redis.io/commands/acl-list

Source code in kvdb/components/session.py

persistence_config = self.settings.persistence.model_dump(exclude_none = True)
persistence_kwargs = self.settings.persistence.extract_kwargs(_prefix = 'persistence_', _exclude_none = True, **self._kwargs)
if persistence_kwargs: persistence_config.update(persistence_kwargs)
if 'name' not in persistence_config: persistence_config['name'] = self.name
if base_key is not None: persistence_config['base_key'] = base_key
persistence_config.update(kwargs)

aacl_load `async`

aacl_load(**kwargs) -> redis.commands.core.ResponseT

Load ACL rules from the configured aclfile.

Note that the server must be configured with the aclfile directive to be able to load ACL rules from an aclfile.

For more information see https://redis.io/commands/acl-load

Source code in kvdb/components/session.py

    """
    Return a Publish/Subscribe object. With this object, you can
    subscribe to channels and listen for messages that get published to
    """
    if retryable is None: retryable = self.settings.retry.pubsub_enabled
    return self.client.pubsub(retryable = retryable, **kwargs)

def apubsub(
    self, 

aacl_log `async`

aacl_log(
    count: typing.Union[int, None] = None, **kwargs
) -> redis.commands.core.ResponseT

Get ACL logs as a list. :param int count: Get logs[0:count]. :rtype: List.

For more information see https://redis.io/commands/acl-log

Source code in kvdb/components/session.py

base_key = persistence_config.get('base_key')
if base_key in self._persistence_ctx:
    return self._persistence_ctx[base_key]
p = KVDBStatefulBackend.as_persistent_dict(
    session = self,
    **persistence_config,
)
self._persistence_ctx[base_key] = p

aacl_log_reset `async`

aacl_log_reset(**kwargs) -> redis.commands.core.ResponseT

Reset ACL logs. :rtype: Boolean.

For more information see https://redis.io/commands/acl-log

Source code in kvdb/components/session.py

    return p

def pubsub(
    self, 
    retryable: Optional[bool] = None,
    **kwargs
) -> PubSubT:

aacl_save `async`

aacl_save(**kwargs) -> redis.commands.core.ResponseT

Save ACL rules to the configured aclfile.

Note that the server must be configured with the aclfile directive to be able to save ACL rules to an aclfile.

For more information see https://redis.io/commands/acl-save

Source code in kvdb/components/session.py

    retryable: Optional[bool] = None,
    **kwargs
) -> AsyncPubSubT:
    """
    Return a Publish/Subscribe object. With this object, you can
    subscribe to channels and listen for messages that get published to
    """
    if retryable is None: retryable = self.settings.retry.pubsub_enabled
    return self.aclient.pubsub(retryable = retryable, **kwargs)

aacl_setuser `async`

aacl_setuser(
    username: str,
    enabled: bool = False,
    nopass: bool = False,
    passwords: typing.Union[
        str, typing.Iterable[str], None
    ] = None,
    hashed_passwords: typing.Union[
        str, typing.Iterable[str], None
    ] = None,
    categories: typing.Optional[
        typing.Iterable[str]
    ] = None,
    commands: typing.Optional[typing.Iterable[str]] = None,
    keys: typing.Optional[
        typing.Iterable[kvdb.types.generic.KeyT]
    ] = None,
    channels: typing.Optional[
        typing.Iterable[redis.typing.ChannelT]
    ] = None,
    selectors: typing.Optional[
        typing.Iterable[
            typing.Tuple[str, kvdb.types.generic.KeyT]
        ]
    ] = None,
    reset: bool = False,
    reset_keys: bool = False,
    reset_channels: bool = False,
    reset_passwords: bool = False,
    **kwargs
) -> redis.commands.core.ResponseT

Create or update an ACL user.

Create or update the ACL for username. If the user already exists, the existing ACL is completely overwritten and replaced with the specified values.

enabled is a boolean indicating whether the user should be allowed to authenticate or not. Defaults to False.

nopass is a boolean indicating whether the can authenticate without a password. This cannot be True if passwords are also specified.

passwords if specified is a list of plain text passwords to add to or remove from the user. Each password must be prefixed with a '+' to add or a '-' to remove. For convenience, the value of passwords can be a simple prefixed string when adding or removing a single password.

hashed_passwords if specified is a list of SHA-256 hashed passwords to add to or remove from the user. Each hashed password must be prefixed with a '+' to add or a '-' to remove. For convenience, the value of hashed_passwords can be a simple prefixed string when adding or removing a single password.

categories if specified is a list of strings representing category permissions. Each string must be prefixed with either a '+' to add the category permission or a '-' to remove the category permission.

commands if specified is a list of strings representing command permissions. Each string must be prefixed with either a '+' to add the command permission or a '-' to remove the command permission.

keys if specified is a list of key patterns to grant the user access to. Keys patterns allow '' to support wildcard matching. For example, '' grants access to all keys while 'cache:*' grants access to all keys that are prefixed with 'cache:'. keys should not be prefixed with a '~'.

reset is a boolean indicating whether the user should be fully reset prior to applying the new ACL. Setting this to True will remove all existing passwords, flags and privileges from the user and then apply the specified rules. If this is False, the user's existing passwords, flags and privileges will be kept and any new specified rules will be applied on top.

reset_keys is a boolean indicating whether the user's key permissions should be reset prior to applying any new key permissions specified in keys. If this is False, the user's existing key permissions will be kept and any new specified key permissions will be applied on top.

reset_channels is a boolean indicating whether the user's channel permissions should be reset prior to applying any new channel permissions specified in channels.If this is False, the user's existing channel permissions will be kept and any new specified channel permissions will be applied on top.

reset_passwords is a boolean indicating whether to remove all existing passwords and the 'nopass' flag from the user prior to applying any new passwords specified in 'passwords' or 'hashed_passwords'. If this is False, the user's existing passwords and 'nopass' status will be kept and any new specified passwords or hashed_passwords will be applied on top.

For more information see https://redis.io/commands/acl-setuser

Source code in kvdb/components/session.py

"""
PubSub Utility Methods
"""

def pipeline(
    self, 
    transaction: Optional[bool] = True, 
    shard_hint: Optional[str] = None, 
    retryable: Optional[bool] = None,
    **kwargs
) -> PipelineT:
    """
    Return a new pipeline object that can queue multiple commands for
    later execution. ``transaction`` indicates whether all commands
    should be executed atomically. Apart from making a group of operations
    atomic, pipelines are useful for reducing the back-and-forth overhead
    between the client and server.
    """
    if retryable is None: retryable = self.settings.retry.pipeline_enabled
    return self.client.pipeline(transaction = transaction, shard_hint = shard_hint, retryable = retryable)

def apipeline(
    self, 
    transaction: Optional[bool] = True, 
    shard_hint: Optional[str] = None, 
    retryable: Optional[bool] = None,
    **kwargs
) -> AsyncPipelineT:
    """
    Return a new pipeline object that can queue multiple commands for
    later execution. ``transaction`` indicates whether all commands
    should be executed atomically. Apart from making a group of operations
    atomic, pipelines are useful for reducing the back-and-forth overhead
    between the client and server.
    """
    if retryable is None: retryable = self.settings.retry.pipeline_enabled
    return self.aclient.pipeline(transaction = transaction, shard_hint = shard_hint, retryable = retryable)

def lock(
    self, 
    name: str, 
    timeout: Optional[Number] = None,
    sleep: Optional[Number] = 0.1,
    blocking: Optional[bool] = True,
    blocking_timeout: Optional[Number] = None,
    thread_local: Optional[bool] = True,
    **kwargs,
) -> Lock:
    """
    Create a new Lock instance named ``name`` using the Redis client
    supplied by ``keydb``.

    ``timeout`` indicates a maximum life for the lock in seconds.
    By default, it will remain locked until release() is called.
    ``timeout`` can be specified as a float or integer, both representing
    the number of seconds to wait.

    ``sleep`` indicates the amount of time to sleep in seconds per loop
    iteration when the lock is in blocking mode and another client is
    currently holding the lock.

    ``blocking`` indicates whether calling ``acquire`` should block until
    the lock has been acquired or to fail immediately, causing ``acquire``
    to return False and the lock not being acquired. Defaults to True.
    Note this value can be overridden by passing a ``blocking``
    argument to ``acquire``.

aacl_users `async`

aacl_users(**kwargs) -> redis.commands.core.ResponseT

Returns a list of all registered users on the server.

For more information see https://redis.io/commands/acl-users

Source code in kvdb/components/session.py

``blocking_timeout`` indicates the maximum amount of time in seconds to
spend trying to acquire the lock. A value of ``None`` indicates
continue trying forever. ``blocking_timeout`` can be specified as a
float or integer, both representing the number of seconds to wait.

aacl_whoami `async`

aacl_whoami(**kwargs) -> redis.commands.core.ResponseT

Get the username for the current connection

For more information see https://redis.io/commands/acl-whoami

Source code in kvdb/components/session.py

``thread_local`` indicates whether the lock token is placed in
thread-local storage. By default, the token is placed in thread local
storage so that a thread only sees its token, not a token set by
another thread. 
"""

aappend `async`

aappend(
    key: kvdb.types.generic.KeyT,
    value: redis.typing.EncodableT,
) -> redis.commands.core.ResponseT

Appends the string value to the value at key. If key doesn't already exist, create it with a value of value. Returns the new length of the value at key.

For more information see https://redis.io/commands/append

aauth `async`

aauth(
    password: str,
    username: typing.Optional[str] = None,
    **kwargs
)

Authenticates the user. If you do not pass username, Redis will try to authenticate for the "default" user. If you do pass username, it will authenticate for the given user. For more information see https://redis.io/commands/auth

Source code in kvdb/components/session.py

        timeout = timeout, 
        sleep = sleep, 
        blocking = blocking, 
        blocking_timeout = blocking_timeout, 
        thread_local = thread_local
    )
if self.state.lock is None: self.state.lock = self.state.locks[name]

abgrewriteaof `async`

abgrewriteaof(**kwargs)

Tell the Redis server to rewrite the AOF file from data in memory.

For more information see https://redis.io/commands/bgrewriteaof

Source code in kvdb/components/session.py

    return self.state.locks[name]

def alock(
    self, 
    name: str, 

abgsave `async`

abgsave(
    schedule: bool = True, **kwargs
) -> redis.commands.core.ResponseT

Tell the Redis server to save its data to disk. Unlike save(), this method is asynchronous and returns immediately.

For more information see https://redis.io/commands/bgsave

Source code in kvdb/components/session.py

    timeout: Optional[Number] = None,
    sleep: Number = 0.1,
    blocking: bool = True,
    blocking_timeout: Optional[Number] = None,
    thread_local: bool = True,
    **kwargs,
) -> AsyncLock:

abitcount `async`

abitcount(
    key: kvdb.types.generic.KeyT,
    start: typing.Union[int, None] = None,
    end: typing.Union[int, None] = None,
    mode: typing.Optional[str] = None,
) -> redis.commands.core.ResponseT

Returns the count of set bits in the value of key. Optional start and end parameters indicate which bytes to consider

For more information see https://redis.io/commands/bitcount

abitfield `async`

abitfield(
    key: kvdb.types.generic.KeyT,
    default_overflow: typing.Union[str, None] = None,
) -> redis.commands.core.BitFieldOperation

Return a BitFieldOperation instance to conveniently construct one or more bitfield operations on key.

For more information see https://redis.io/commands/bitfield

abitfield_ro `async`

abitfield_ro(
    key: kvdb.types.generic.KeyT,
    encoding: str,
    offset: redis.typing.BitfieldOffsetT,
    items: typing.Optional[list] = None,
) -> redis.commands.core.ResponseT

Return an array of the specified bitfield values where the first value is found using encoding and offset parameters and remaining values are result of corresponding encoding/offset pairs in optional list items Read-only variant of the BITFIELD command.

For more information see https://redis.io/commands/bitfield_ro

abitop `async`

abitop(
    operation: str,
    dest: kvdb.types.generic.KeyT,
    *keys: kvdb.types.generic.KeyT
) -> redis.commands.core.ResponseT

Perform a bitwise operation using operation between keys and store the result in dest.

For more information see https://redis.io/commands/bitop

abitpos `async`

abitpos(
    key: kvdb.types.generic.KeyT,
    bit: int,
    start: typing.Union[int, None] = None,
    end: typing.Union[int, None] = None,
    mode: typing.Optional[str] = None,
) -> redis.commands.core.ResponseT

Return the position of the first bit set to 1 or 0 in a string. start and end defines search range. The range is interpreted as a range of bytes and not a range of bits, so start=0 and end=2 means to look at the first three bytes.

For more information see https://redis.io/commands/bitpos

ablmove `async`

ablmove(
    first_list: str,
    second_list: str,
    timeout: int,
    src: str = "LEFT",
    dest: str = "RIGHT",
) -> redis.commands.core.ResponseT

Blocking version of lmove.

For more information see https://redis.io/commands/blmove

ablmpop `async`

ablmpop(
    timeout: float,
    numkeys: int,
    *args: typing.List[str],
    direction: str,
    count: typing.Optional[int] = 1
) -> typing.Optional[list]

Pop count values (default 1) from first non-empty in the list of provided key names.

When all lists are empty this command blocks the connection until another client pushes to it or until the timeout, timeout of 0 blocks indefinitely

For more information see https://redis.io/commands/blmpop

ablpop `async`

ablpop(
    keys: typing.List, timeout: typing.Optional[int] = 0
) -> typing.Union[typing.Awaitable[list], list]

LPOP a value off of the first non-empty list named in the keys list.

If none of the lists in keys has a value to LPOP, then block for timeout seconds, or until a value gets pushed on to one of the lists.

If timeout is 0, then block indefinitely.

For more information see https://redis.io/commands/blpop

abrpop `async`

abrpop(
    keys: typing.List, timeout: typing.Optional[int] = 0
) -> typing.Union[typing.Awaitable[list], list]

RPOP a value off of the first non-empty list named in the keys list.

If none of the lists in keys has a value to RPOP, then block for timeout seconds, or until a value gets pushed on to one of the lists.

If timeout is 0, then block indefinitely.

For more information see https://redis.io/commands/brpop

abrpoplpush `async`

abrpoplpush(
    src: str, dst: str, timeout: typing.Optional[int] = 0
) -> typing.Union[
    typing.Awaitable[typing.Optional[str]],
    typing.Optional[str],
]

Pop a value off the tail of src, push it on the head of dst and then return it.

This command blocks until a value is in src or until timeout seconds elapse, whichever is first. A timeout value of 0 blocks forever.

For more information see https://redis.io/commands/brpoplpush

abzmpop `async`

abzmpop(
    timeout: float,
    numkeys: int,
    keys: typing.List[str],
    min: typing.Optional[bool] = False,
    max: typing.Optional[bool] = False,
    count: typing.Optional[int] = 1,
) -> typing.Optional[list]

Pop count values (default 1) off of the first non-empty sorted set named in the keys list.

If none of the sorted sets in keys has a value to pop, then block for timeout seconds, or until a member gets added to one of the sorted sets.

If timeout is 0, then block indefinitely.

For more information see https://redis.io/commands/bzmpop

abzpopmax `async`

abzpopmax(
    keys: redis.typing.KeysT,
    timeout: redis.typing.TimeoutSecT = 0,
) -> redis.commands.core.ResponseT

ZPOPMAX a value off of the first non-empty sorted set named in the keys list.

If none of the sorted sets in keys has a value to ZPOPMAX, then block for timeout seconds, or until a member gets added to one of the sorted sets.

If timeout is 0, then block indefinitely.

For more information see https://redis.io/commands/bzpopmax

abzpopmin `async`

abzpopmin(
    keys: redis.typing.KeysT,
    timeout: redis.typing.TimeoutSecT = 0,
) -> redis.commands.core.ResponseT

ZPOPMIN a value off of the first non-empty sorted set named in the keys list.

If none of the sorted sets in keys has a value to ZPOPMIN, then block for timeout seconds, or until a member gets added to one of the sorted sets.

If timeout is 0, then block indefinitely.

For more information see https://redis.io/commands/bzpopmin

acl_cat

acl_cat(
    category: typing.Union[str, None] = None, **kwargs
) -> redis.commands.core.ResponseT

Returns a list of categories or commands within a category.

If category is not supplied, returns a list of all categories. If category is supplied, returns a list of all commands within that category.

For more information see https://redis.io/commands/acl-cat

acl_deluser

acl_deluser(
    *username: str, **kwargs
) -> redis.commands.core.ResponseT

Delete the ACL for the specified usernames

For more information see https://redis.io/commands/acl-deluser

acl_dryrun

acl_dryrun(username, *args, **kwargs)

Simulate the execution of a given command by a given username.

For more information see https://redis.io/commands/acl-dryrun

acl_genpass

acl_genpass(
    bits: typing.Union[int, None] = None, **kwargs
) -> redis.commands.core.ResponseT

Generate a random password value. If bits is supplied then use this number of bits, rounded to the next multiple of 4. See: https://redis.io/commands/acl-genpass

acl_getuser

acl_getuser(
    username: str, **kwargs
) -> redis.commands.core.ResponseT

Get the ACL details for the specified username.

If username does not exist, return None

For more information see https://redis.io/commands/acl-getuser

acl_help

acl_help(**kwargs) -> redis.commands.core.ResponseT

The ACL HELP command returns helpful text describing the different subcommands.

For more information see https://redis.io/commands/acl-help

acl_list

acl_list(**kwargs) -> redis.commands.core.ResponseT

Return a list of all ACLs on the server

For more information see https://redis.io/commands/acl-list

acl_load

acl_load(**kwargs) -> redis.commands.core.ResponseT

Load ACL rules from the configured aclfile.

Note that the server must be configured with the aclfile directive to be able to load ACL rules from an aclfile.

For more information see https://redis.io/commands/acl-load

acl_log

acl_log(
    count: typing.Union[int, None] = None, **kwargs
) -> redis.commands.core.ResponseT

Get ACL logs as a list. :param int count: Get logs[0:count]. :rtype: List.

For more information see https://redis.io/commands/acl-log

acl_log_reset

acl_log_reset(**kwargs) -> redis.commands.core.ResponseT

Reset ACL logs. :rtype: Boolean.

For more information see https://redis.io/commands/acl-log

acl_save

acl_save(**kwargs) -> redis.commands.core.ResponseT

Save ACL rules to the configured aclfile.

Note that the server must be configured with the aclfile directive to be able to save ACL rules to an aclfile.

For more information see https://redis.io/commands/acl-save

acl_setuser

acl_setuser(
    username: str,
    enabled: bool = False,
    nopass: bool = False,
    passwords: typing.Union[
        str, typing.Iterable[str], None
    ] = None,
    hashed_passwords: typing.Union[
        str, typing.Iterable[str], None
    ] = None,
    categories: typing.Optional[
        typing.Iterable[str]
    ] = None,
    commands: typing.Optional[typing.Iterable[str]] = None,
    keys: typing.Optional[
        typing.Iterable[kvdb.types.generic.KeyT]
    ] = None,
    channels: typing.Optional[
        typing.Iterable[redis.typing.ChannelT]
    ] = None,
    selectors: typing.Optional[
        typing.Iterable[
            typing.Tuple[str, kvdb.types.generic.KeyT]
        ]
    ] = None,
    reset: bool = False,
    reset_keys: bool = False,
    reset_channels: bool = False,
    reset_passwords: bool = False,
    **kwargs
) -> redis.commands.core.ResponseT