glide/commands/
generic.rs

1// Copyright Valkey GLIDE Project Contributors - SPDX Identifier: Apache-2.0
2//! Generic (key) commands. Mirrors Python's generic command surface.
3
4use crate::ValkeyResult;
5use crate::cmd::Cmd;
6use crate::commands::options::{Limit, MigrateOptions, OrderBy, RestoreOptions};
7use crate::executor::CommandExecutor;
8use crate::value::FromValkeyValue;
9use crate::value::ValkeyValue;
10use crate::value::to_glide_error;
11use crate::write::ToValkeyArgs;
12use async_trait::async_trait;
13use bytes::Bytes;
14
15/// Generic key-space commands (`DEL`, `EXISTS`, `EXPIRE`, `TTL`, `RENAME`, ...).
16#[async_trait]
17pub trait GenericCommands: CommandExecutor {
18    // TODO #7082: add `expire`/`pexpire`/`expire_at`/`pexpire_at` variants that take
19    // the NX/XX/GT/LT condition options (the `ExpireOption` type already exists).
20
21    /// Return a random key from the keyspace (`RANDOMKEY`).
22    async fn randomkey(&self) -> ValkeyResult<Option<Bytes>> {
23        let mut cmd = Cmd::new();
24        cmd.arg("RANDOMKEY");
25        Option::<Bytes>::from_owned_valkey_value(self.execute_command(cmd, None).await?)
26    }
27
28    /// Serialize `key` (`DUMP`). Returns `None` if the key does not exist.
29    async fn dump<K: ToValkeyArgs + Send>(&self, key: K) -> ValkeyResult<Option<Bytes>> {
30        let mut cmd = Cmd::new();
31        cmd.arg("DUMP").arg(key);
32        Option::<Bytes>::from_owned_valkey_value(self.execute_command(cmd, None).await?)
33    }
34
35    /// Touch the given keys, returning how many were touched (`TOUCH`).
36    async fn touch<K: ToValkeyArgs + Send + Sync>(&self, keys: &[K]) -> ValkeyResult<i64> {
37        let mut cmd = Cmd::new();
38        cmd.arg("TOUCH");
39        for k in keys {
40            cmd.arg(k);
41        }
42        i64::from_owned_valkey_value(self.execute_command(cmd, None).await?)
43    }
44
45    /// Sort the elements at `key` (`SORT`), optionally by order and with an
46    /// optional `LIMIT offset count`.
47    async fn sort<K: ToValkeyArgs + Send>(
48        &self,
49        key: K,
50        order: Option<OrderBy>,
51        limit: Option<Limit>,
52        alpha: bool,
53    ) -> ValkeyResult<Vec<Bytes>> {
54        let mut cmd = Cmd::new();
55        cmd.arg("SORT").arg(key);
56        if let Some(l) = limit {
57            cmd.arg("LIMIT").arg(l.offset).arg(l.count);
58        }
59        if let Some(o) = order {
60            cmd.arg(o.as_arg());
61        }
62        if alpha {
63            cmd.arg("ALPHA");
64        }
65        match self.execute_command(cmd, None).await? {
66            ValkeyValue::Array(items) => items
67                .into_iter()
68                .map(Bytes::from_owned_valkey_value)
69                .collect(),
70            ValkeyValue::Nil => Ok(Vec::new()),
71            other => Ok(vec![Bytes::from_owned_valkey_value(other)?]),
72        }
73    }
74
75    /// Create a key from a serialized payload produced by `DUMP` (`RESTORE`).
76    async fn restore<K: ToValkeyArgs + Send, V: ToValkeyArgs + Send>(
77        &self,
78        key: K,
79        ttl_ms: i64,
80        serialized: V,
81        options: RestoreOptions,
82    ) -> ValkeyResult<()> {
83        let mut cmd = Cmd::new();
84        cmd.arg("RESTORE")
85            .arg(key)
86            .arg(ttl_ms)
87            .arg(serialized)
88            .arg(options);
89        <()>::from_owned_valkey_value(self.execute_command(cmd, None).await?)
90    }
91
92    /// Block until `numreplicas` replicas acknowledge previous writes, or until
93    /// `timeout_ms` elapses (`WAIT`). Returns the number of replicas reached.
94    async fn wait(&self, numreplicas: i64, timeout_ms: i64) -> ValkeyResult<i64> {
95        let mut cmd = Cmd::new();
96        cmd.arg("WAIT").arg(numreplicas).arg(timeout_ms);
97        i64::from_owned_valkey_value(self.execute_command(cmd, None).await?)
98    }
99
100    /// Sort the elements at `key` and store the result into `destination`
101    /// (`SORT ... STORE destination`). Returns the number of elements stored.
102    async fn sort_store<K: ToValkeyArgs + Send, D: ToValkeyArgs + Send>(
103        &self,
104        key: K,
105        destination: D,
106        order: Option<OrderBy>,
107        limit: Option<Limit>,
108        alpha: bool,
109    ) -> ValkeyResult<i64> {
110        let mut cmd = Cmd::new();
111        cmd.arg("SORT").arg(key);
112        if let Some(l) = limit {
113            cmd.arg("LIMIT").arg(l.offset).arg(l.count);
114        }
115        if let Some(o) = order {
116            cmd.arg(o.as_arg());
117        }
118        if alpha {
119            cmd.arg("ALPHA");
120        }
121        cmd.arg("STORE").arg(destination);
122        i64::from_owned_valkey_value(self.execute_command(cmd, None).await?)
123    }
124
125    /// Read-only variant of `SORT` (`SORT_RO`); returns the sorted elements.
126    async fn sort_ro<K: ToValkeyArgs + Send>(
127        &self,
128        key: K,
129        order: Option<OrderBy>,
130        limit: Option<Limit>,
131        alpha: bool,
132    ) -> ValkeyResult<Vec<Bytes>> {
133        let mut cmd = Cmd::new();
134        cmd.arg("SORT_RO").arg(key);
135        if let Some(l) = limit {
136            cmd.arg("LIMIT").arg(l.offset).arg(l.count);
137        }
138        if let Some(o) = order {
139            cmd.arg(o.as_arg());
140        }
141        if alpha {
142            cmd.arg("ALPHA");
143        }
144        match self.execute_command(cmd, None).await? {
145            ValkeyValue::Array(items) => items
146                .into_iter()
147                .map(Bytes::from_owned_valkey_value)
148                .collect(),
149            ValkeyValue::Nil => Ok(Vec::new()),
150            other => Ok(vec![Bytes::from_owned_valkey_value(other)?]),
151        }
152    }
153
154    /// Move `key` to another logical database (`MOVE`). Returns whether the key
155    /// was moved.
156    async fn move_key<K: ToValkeyArgs + Send>(&self, key: K, db: i64) -> ValkeyResult<bool> {
157        let mut cmd = Cmd::new();
158        cmd.arg("MOVE").arg(key).arg(db);
159        bool::from_owned_valkey_value(self.execute_command(cmd, None).await?)
160    }
161
162    /// Atomically transfer a key to another instance (`MIGRATE`).
163    async fn migrate<H: ToValkeyArgs + Send, K: ToValkeyArgs + Send>(
164        &self,
165        host: H,
166        port: i64,
167        key: K,
168        destination_db: i64,
169        timeout_ms: i64,
170        options: MigrateOptions,
171    ) -> ValkeyResult<()> {
172        let mut cmd = Cmd::new();
173        cmd.arg("MIGRATE")
174            .arg(host)
175            .arg(port)
176            .arg(key)
177            .arg(destination_db)
178            .arg(timeout_ms)
179            .arg(options);
180        <()>::from_owned_valkey_value(self.execute_command(cmd, None).await?)
181    }
182
183    /// Watch the given keys for changes before a transaction (`WATCH`).
184    ///
185    /// Note: `WATCH` is connection-scoped, but GLIDE multiplexes commands over
186    /// shared connections. The optimistic lock is therefore only reliable while
187    /// the caller has exclusive use of the client through `EXEC`/`UNWATCH`.
188    /// Dedicated scoped connections are tracked in [#6917].
189    ///
190    /// [#6917]: https://github.com/valkey-io/valkey-glide/issues/6917
191    async fn watch<K: ToValkeyArgs + Send + Sync>(&self, keys: &[K]) -> ValkeyResult<()> {
192        let mut cmd = Cmd::new();
193        cmd.arg("WATCH");
194        for k in keys {
195            cmd.arg(k);
196        }
197        <()>::from_owned_valkey_value(self.execute_command(cmd, None).await?)
198    }
199
200    /// Forget all watched keys (`UNWATCH`).
201    ///
202    /// See [`watch`](Self::watch) for the connection-scoping caveat that
203    /// applies to `WATCH`/`UNWATCH` on GLIDE's multiplexed client.
204    async fn unwatch(&self) -> ValkeyResult<()> {
205        let mut cmd = Cmd::new();
206        cmd.arg("UNWATCH");
207        <()>::from_owned_valkey_value(self.execute_command(cmd, None).await?)
208    }
209}
210
211pub(crate) fn parse_scan_reply(reply: ValkeyValue) -> ValkeyResult<(String, Vec<Bytes>)> {
212    match reply {
213        ValkeyValue::Array(mut items) if items.len() == 2 => {
214            let keys_val = items.pop().unwrap();
215            let cursor_val = items.pop().unwrap();
216            let cursor = String::from_owned_valkey_value(cursor_val)?;
217            let keys = match keys_val {
218                ValkeyValue::Array(elems) => elems
219                    .into_iter()
220                    .map(Bytes::from_owned_valkey_value)
221                    .collect::<ValkeyResult<Vec<_>>>()?,
222                ValkeyValue::Nil => Vec::new(),
223                other => {
224                    return Err(to_glide_error(other, "Unexpected SCAN keys shape."));
225                }
226            };
227            Ok((cursor, keys))
228        }
229        other => Err(to_glide_error(other, "Unexpected SCAN reply shape.")),
230    }
231}
232
233impl<T: CommandExecutor + ?Sized> GenericCommands for T {}
glide/commands/generic.rs

glide/commands/
generic.rs
