feat: add data vault store with key-value operations

- DataVaultStore class with sync get/set/delete/list_keys methods
- AsyncDataVaultStore class with async variants
- Export both from models package
This commit is contained in:
Ejiro Asiuwhu 2026-03-25 01:51:59 +01:00
parent c5ab83984f
commit 299718c9db
2 changed files with 90 additions and 0 deletions

View File

@ -1,5 +1,6 @@
"""Stack Auth data models -- re-exports all model classes."""
from stack_auth.models.data_vault import AsyncDataVaultStore, DataVaultStore
from stack_auth.models.api_keys import (
ApiKey,
TeamApiKey,
@ -19,6 +20,8 @@ from stack_auth.models.teams import ServerTeam, Team, TeamInvitation, TeamMember
from stack_auth.models.users import BaseUser, ServerUser
__all__ = [
"DataVaultStore",
"AsyncDataVaultStore",
"BaseUser",
"ServerUser",
"Team",

View File

@ -0,0 +1,87 @@
"""Data vault store classes for server-side key-value storage."""
from __future__ import annotations
from typing import Any
from stack_auth.errors import NotFoundError
class DataVaultStore:
"""Synchronous data vault store -- a server-side key-value store.
Obtained via ``StackServerApp.get_data_vault_store(store_id)``.
"""
def __init__(self, store_id: str, *, _client: Any) -> None:
self.id = store_id
self._client = _client
self._base_path = f"/data-vault/stores/{store_id}/items"
def get(self, key: str) -> str | None:
"""Get the value for a key, or ``None`` if not found."""
try:
data = self._client.request("GET", f"{self._base_path}/{key}")
except NotFoundError:
return None
if data is None:
return None
return data.get("value") if isinstance(data, dict) else data
def set(self, key: str, value: str) -> None:
"""Set or update the value for a key."""
self._client.request("PUT", f"{self._base_path}/{key}", body={"value": value})
def delete(self, key: str) -> None:
"""Delete a key-value pair. No error if key doesn't exist."""
self._client.request("DELETE", f"{self._base_path}/{key}")
def list_keys(self) -> list[str]:
"""Return all keys in the store."""
data = self._client.request("GET", self._base_path)
if data is None:
return []
# Response may be {"items": [...]} or a list directly
if isinstance(data, dict):
return data.get("items", [])
return data
class AsyncDataVaultStore:
"""Asynchronous data vault store -- a server-side key-value store.
Obtained via ``AsyncStackServerApp.get_data_vault_store(store_id)``.
"""
def __init__(self, store_id: str, *, _client: Any) -> None:
self.id = store_id
self._client = _client
self._base_path = f"/data-vault/stores/{store_id}/items"
async def get(self, key: str) -> str | None:
"""Get the value for a key, or ``None`` if not found."""
try:
data = await self._client.request("GET", f"{self._base_path}/{key}")
except NotFoundError:
return None
if data is None:
return None
return data.get("value") if isinstance(data, dict) else data
async def set(self, key: str, value: str) -> None:
"""Set or update the value for a key."""
await self._client.request("PUT", f"{self._base_path}/{key}", body={"value": value})
async def delete(self, key: str) -> None:
"""Delete a key-value pair. No error if key doesn't exist."""
await self._client.request("DELETE", f"{self._base_path}/{key}")
async def list_keys(self) -> list[str]:
"""Return all keys in the store."""
data = await self._client.request("GET", self._base_path)
if data is None:
return []
# Response may be {"items": [...]} or a list directly
if isinstance(data, dict):
return data.get("items", [])
return data