sf-queue-sdk 0.3.1

Python SDK for sf-queue - Redis-based queue system

sf-queue-sdk

Python SDK for sf-queue. Enqueues emails via Redis Streams with optional blocking confirmation from the Go consumer service.

Installation

pip install sf-queue-sdk

Setup

from queue_sdk import QueueClient

client = QueueClient(
    redis_url="redis://localhost:6379",
    redis_password="your-password",
    environment="staging",  # prefixes stream names: staging:{email}
)

Single Email

Fire and forget

Enqueues the email and returns immediately. Does not wait for the consumer to process it.

result = client.email.send(
    to="[e-mail]",
    preview="Welcome to StudyFetch!",
    subject="Welcome to StudyFetch!",
    paragraphs=[
        "Hey there,",
        "Welcome to the StudyFetch community!",
        "Thanks for joining us.",
    ],
    button={
        "text": "Go to Platform",
        "href": "https://www.studyfetch.com/platform",
    },
)

print("Enqueued:", result.message_id)

Send and wait for confirmation

Enqueues the email and blocks until the Go consumer processes it (or timeout).

result = client.email.send_and_wait(
    to="[e-mail]",
    preview="Reset your password",
    subject="StudyFetch: Reset Your Password",
    paragraphs=[
        "Hi There,",
        "Click the button below to reset your password.",
    ],
    button={
        "text": "Reset Password",
        "href": "https://www.studyfetch.com/reset?token=abc",
    },
    timeout=30,  # optional, default 30s
)

print(result.success)     # True or False
print(result.message_id)  # request ID
print(result.error)       # error message if failed

Batch Email

Send the same email content to multiple recipients (up to 100). The Go consumer sends to each recipient individually.

Fire and forget

result = client.email.send_batch(
    to=[
        "[e-mail]",
        "[e-mail]",
        "[e-mail]",
    ],
    preview="You have been invited to join a class!",
    subject="StudyFetch: Class Invitation",
    paragraphs=[
        "Hi There,",
        'You have been invited to join "Intro to CS" on StudyFetch!',
        "Click the button below to accept the invite.",
    ],
    button={
        "text": "Accept Invite",
        "href": "https://www.studyfetch.com/invite/abc",
    },
)

print("Enqueued:", result.message_id)

Send and wait for confirmation

result = client.email.send_batch_and_wait(
    to=[
        "[e-mail]",
        "[e-mail]",
        "[e-mail]",
    ],
    preview="You have been invited to join a class!",
    subject="StudyFetch: Class Invitation",
    paragraphs=[
        "Hi There,",
        'You have been invited to join "Intro to CS" on StudyFetch!',
        "Click the button below to accept the invite.",
    ],
    button={
        "text": "Accept Invite",
        "href": "https://www.studyfetch.com/invite/abc",
    },
    timeout=30,
)

print(result.success)     # True if at least some sent
print(result.message_id)  # request ID
print(result.total)       # 3
print(result.successful)  # number sent successfully
print(result.failed)      # number that failed
print(result.error)       # error message if all failed

All Email Fields

Field	Type	Required	Description
to	str	Yes (single)	Recipient email address
to	list[str]	Yes (batch)	List of recipient emails (max 100)
preview	str	Yes	Preview text shown in email clients
subject	str	Yes	Email subject line
paragraphs	list[str]	Yes	Body content as paragraph strings
button	{"text": str, "href": str}	No	Call-to-action button
reply_to	str	No	Reply-to email address
image	{"src": str, "alt"?: str, "width"?: int, "height"?: int}	No	Image in email body
service	str	No	Service name for routing (e.g. "sparkles"). Routes to service-specific credentials, from address, and template. Defaults to the base config when omitted.
template	str	No	Template name override (e.g. "sparkles"). Overrides the service's default template. Falls back to "studyfetch" when omitted.

Optional Fields

# With all optional fields
client.email.send(
    to="[e-mail]",
    preview="Support Request",
    subject="StudyFetch: Support Request",
    paragraphs=["Hi There,", "You received a support request.", issue],
    reply_to="[e-mail]",
    image={
        "src": "https://example.com/logo.png",
        "alt": "Logo",
        "width": 150,
        "height": 50,
    },
)

Multi-Service Routing

Use service to route emails through a different service's credentials and branding. The consumer resolves the service name to its configured Resend API key, from address, and default template. Use template to override the template for a specific email.

# Send via a different service's email account and template
client.email.send(
    to="[e-mail]",
    preview="Welcome!",
    subject="Welcome!",
    paragraphs=["Welcome to the platform!"],
    service="sparkles",
)

# Override the template for a specific email
client.email.send(
    to="[e-mail]",
    preview="Welcome!",
    subject="Welcome!",
    paragraphs=["Welcome to the platform!"],
    service="sparkles",
    template="sparkles",
)

Migrating from sendEmail / sendBatchEmail

The SDK is a drop-in replacement. Field names match the existing functions:

# BEFORE
send_email(to=to, preview=preview, subject=subject, paragraphs=paragraphs, button=button)

# AFTER (fire and forget)
client.email.send(to=to, preview=preview, subject=subject, paragraphs=paragraphs, button=button)

# AFTER (wait for confirmation)
client.email.send_and_wait(to=to, preview=preview, subject=subject, paragraphs=paragraphs, button=button)

# BEFORE (batch)
send_batch_email(to=[...], preview=preview, subject=subject, paragraphs=paragraphs, button=button)

# AFTER (batch, fire and forget)
client.email.send_batch(to=[...], preview=preview, subject=subject, paragraphs=paragraphs, button=button)

# AFTER (batch, wait for confirmation)
client.email.send_batch_and_wait(to=[...], preview=preview, subject=subject, paragraphs=paragraphs, button=button)

Methods and Response Types

Method	Return Type	Fields
send()	SendResult	message_id
send_and_wait()	EmailResponse	success, message_id, error?, processed_at?
send_batch()	SendResult	message_id
send_batch_and_wait()	BatchEmailResponse	success, message_id, error?, processed_at?, total, successful, failed

Topic Assignment

Fire and forget

result = client.topic_assignment.send(
    topic_vector_id="507f1f77bcf86cd799439011",
    topic_id="topic_abc123",
    user_id="user_xyz",
    current_assignment=None,
    new_assignment="faiss_1422.0",
)
print("Enqueued:", result.message_id)

Send and wait for confirmation

result = client.topic_assignment.send_and_wait(
    topic_vector_id="507f1f77bcf86cd799439011",
    topic_id="topic_abc123",
    user_id="user_xyz",
    current_assignment=None,
    new_assignment="faiss_1422.0",
    timeout=30,
)
print(result.success)
print(result.message_id)

Batch (fire and forget)

result = client.topic_assignment.send_batch(
    assignments=[
        {
            "topic_vector_id": "507f1f77bcf86cd799439011",
            "topic_id": "topic_abc123",
            "user_id": "user_xyz",
            "current_assignment": None,
            "new_assignment": "faiss_1422.0",
        },
        {
            "topic_vector_id": "507f1f77bcf86cd799439012",
            "topic_id": "topic_def456",
            "user_id": "user_xyz",
            "current_assignment": "faiss_1422",
            "new_assignment": "faiss_1422.1",
        },
    ],
)

Batch (wait for confirmation)

result = client.topic_assignment.send_batch_and_wait(
    assignments=[
        {
            "topic_vector_id": "507f1f77bcf86cd799439011",
            "topic_id": "topic_abc123",
            "user_id": "user_xyz",
            "current_assignment": None,
            "new_assignment": "faiss_1422.0",
        },
    ],
    timeout=30,
)
print(result.success)
print(result.total)
print(result.successful)
print(result.failed)

Topic Assignment Fields

Field	Type	Required	Description
topic_vector_id	str	Yes	MongoDB _id of the topic_vector
topic_id	str	Yes	The topicId field
user_id	str	Yes	The userId who owns the topic
current_assignment	Optional[str]	Yes	Current cluster_id (None for first-time)
new_assignment	str	Yes	The new cluster_id

Tool Assignment

Fire and forget

result = client.tool_assignment.send(
    tool_id="507f1f77bcf86cd799439011",
    user_id="user_xyz",
    topic_id="topic_abc123",
    current_assignment=None,
    new_assignment="faiss_1422.0",
)

Send and wait for confirmation

result = client.tool_assignment.send_and_wait(
    tool_id="507f1f77bcf86cd799439011",
    user_id="user_xyz",
    topic_id="topic_abc123",
    current_assignment=None,
    new_assignment="faiss_1422.0",
    timeout=30,
)

Batch (fire and forget)

result = client.tool_assignment.send_batch(
    assignments=[
        {
            "tool_id": "507f1f77bcf86cd799439011",
            "user_id": "user_xyz",
            "topic_id": "topic_abc123",
            "current_assignment": None,
            "new_assignment": "faiss_1422.0",
        },
        {
            "tool_id": "507f1f77bcf86cd799439012",
            "user_id": "user_xyz",
            "topic_id": "topic_abc123",
            "current_assignment": "faiss_1422",
            "new_assignment": "faiss_1422.1",
        },
    ],
)

Batch (wait for confirmation)

result = client.tool_assignment.send_batch_and_wait(
    assignments=[
        {"tool_id": "...", "user_id": "...", "topic_id": "...", "current_assignment": None, "new_assignment": "..."},
    ],
    timeout=30,
)

Tool Assignment Fields

Field	Type	Required	Description
tool_id	str	Yes	MongoDB _id of the tool document
user_id	str	Yes	The userId who owns the tool
topic_id	str	Yes	The topicId the tool belongs to
current_assignment	Optional[str]	Yes	Current cluster_id (None for new embeddings)
new_assignment	str	Yes	The cluster_id to assign to

Recommendations

Fire and forget

result = client.recommendations.create(
    user_id="user_xyz",
    topic_id="topic_abc123",
    limit=10,  # optional
)
print("Enqueued:", result.message_id)

Create and wait for result

result = client.recommendations.create_and_wait(
    user_id="user_xyz",
    topic_id="topic_abc123",
    limit=10,  # optional
    timeout=30,
)
print(result.success)
print(result.message_id)

Batch (fire and forget)

result = client.recommendations.create_batch([
    {"user_id": "user_1", "topic_id": "topic_1", "limit": 10},
    {"user_id": "user_2", "topic_id": "topic_2"},
])
print("Enqueued:", result.message_id)

Batch (wait for confirmation)

result = client.recommendations.create_batch_and_wait(
    [
        {"user_id": "user_1", "topic_id": "topic_1"},
        {"user_id": "user_2", "topic_id": "topic_2"},
    ],
    timeout=60,
)
print(result.success)
print(result.total)
print(result.successful)
print(result.failed)

Recommendation Fields

Field	Type	Required	Description
user_id	str	Yes	User requesting recommendations
topic_id	str	Yes	Topic context for recommendations
limit	Optional[int]	No	Max recommendations to return

PDF to Image Conversion

Convert PDFs, videos, and other documents to images. The Go pdf-to-image-queue worker downloads the source file, converts it, uploads the result to storage, and returns the image URL.

Fire and forget

result = client.pdf_to_image.convert(
    url="https://example.com/document.pdf",
    material_id="mat_abc123",
    # page=1,        # optional
    # format="webp",  # optional
    # quality=80,     # optional
)

print("Enqueued:", result.message_id)

Convert and wait for result

result = client.pdf_to_image.convert_and_wait(
    url="https://example.com/document.pdf",
    material_id="mat_abc123",
    # page=1,        # optional
    # format="webp",  # optional
    # quality=80,     # optional
    # timeout=30,     # optional
)

print(result.success)    # True or False
print(result.image_url)  # URL of the converted image
print(result.format)     # "webp" or "png"
print(result.filename)   # output filename
print(result.error)      # error message if failed

Convert Fields

Field	Type	Required	Description
url	str	Yes	URL of the source file (PDF, video, etc.)
material_id	str	Yes	ID of the material to associate with the conversion
page	Optional[int]	No	PDF page number to convert (default: 1, must be a positive integer)
format	Optional[str]	No	Output image format: "png" or "webp" (default: "webp")
quality	Optional[int]	No	Image quality 1-100 (default: 80, only applies to webp)

Convert Response

Field	Type	Description
success	bool	Whether the conversion succeeded
message_id	str	Request ID
image_url	Optional[str]	URL of the uploaded image
format	Optional[str]	Output format used
filename	Optional[str]	Output filename
error	Optional[str]	Error message if success is False
processed_at	Optional[str]	ISO timestamp of when the worker processed the request

Document Upload

Enqueue document processing jobs for the Python document-upload-queue worker. The worker runs a multi-step pipeline: PDF chunking, embedding, clustering, LLM title generation, and study plan assembly. This is fire-and-forget only -- the worker processes asynchronously and pushes results to S3.

Fire and forget

result = client.document_upload.send(
    uploaded_file_id="file_abc123",
    study_set_id="set_xyz",
    user_id="user_123",
    text_extract_s3_key="uploadedfiles/file_abc123/textContent",
    pdf_s3_key="uploads/document.pdf",
    file_type="application/pdf",
    effective_file_type="application/pdf",
    filename="Chapter 5 - Thermodynamics.pdf",
    material_title="Thermodynamics",
    url="https://storage.example.com/uploads/document.pdf",
    generate_notes=True,
    generate_study_path=True,
)

print("Enqueued:", result.message_id)

With optional fields

result = client.document_upload.send(
    uploaded_file_id="file_abc123",
    study_set_id="set_xyz",
    user_id="user_123",
    text_extract_s3_key="uploadedfiles/file_abc123/textContent",
    pdf_s3_key="uploads/document.pdf",
    file_type="application/pdf",
    effective_file_type="application/pdf",
    filename="Chapter 5 - Thermodynamics.pdf",
    material_title="Thermodynamics",
    url="https://storage.example.com/uploads/document.pdf",
    generate_notes=True,
    generate_study_path=True,
    splitting="auto",
    language="en",
    use_markdown_ocr=False,
    flashcard_amount=12,
    skill_level="undergraduate",
    typeof_notes="comprehensive",
    material_group_id="group_123",
    folder_id="folder_456",
    upload_batch_id="batch_789",
)

Document Upload Fields

Field	Type	Required	Description
uploaded_file_id	str	Yes	ID of the uploaded file record
study_set_id	str	Yes	Study set to associate the processed material with
user_id	str	Yes	User who uploaded the document
text_extract_s3_key	str	Yes	S3 key for extracted text content
pdf_s3_key	str	Yes	S3 key of the source PDF
file_type	str	Yes	MIME type of the uploaded file
effective_file_type	str	Yes	Resolved file type after any conversion
filename	str	Yes	Original filename
material_title	str	Yes	Display title for the material
url	str	Yes	Download URL for the source file
generate_notes	bool	Yes	Whether to generate study notes
generate_study_path	bool	Yes	Whether to generate a study path
splitting	str	No	Splitting strategy (default: "auto")
language	str	No	Target language code (default: "en")
use_markdown_ocr	bool	No	Use Markdown OCR for image interpretation (default: False)
flashcard_amount	int	No	Number of flashcards to generate (default: 12)
new_url	Optional[str]	No	Alternative URL if file was re-uploaded
typeof_notes	Optional[str]	No	Note generation mode ("comprehensive", "detailed", "summarized")
skill_level	Optional[str]	No	User's skill level for content adaptation
features	Optional[dict]	No	Feature flags for optional processing steps
selected_questions	Optional[list]	No	Pre-selected questions for the material
material_group_id	Optional[str]	No	Material group for organization
folder_id	Optional[str]	No	Folder for organization
upload_batch_id	Optional[str]	No	Batch ID when multiple files uploaded together
youtube_title	Optional[str]	No	Title override for YouTube video transcripts
skip_material_creation	Optional[bool]	No	Skip folder and material creation when the material already exists (e.g. live lectures)
existing_material_id	Optional[str]	No	ID of an existing material to link to topics instead of creating new ones (used with skip_material_creation)

Response Type

Method	Return Type	Fields
send()	SendResult	message_id
get_status(request_id)	DocumentUploadStatus	See below

Status Tracking

After enqueueing a document upload, use get_status() to poll the processing progress. The send() method returns a message_id that you pass to get_status().

result = client.document_upload.send(
    uploaded_file_id="file_abc123",
    study_set_id="set_xyz",
    user_id="user_123",
    text_extract_s3_key="uploadedfiles/file_abc123/textContent",
    pdf_s3_key="uploads/document.pdf",
    file_type="application/pdf",
    effective_file_type="application/pdf",
    filename="Chapter 5 - Thermodynamics.pdf",
    material_title="Thermodynamics",
    url="https://storage.example.com/uploads/document.pdf",
    generate_notes=True,
    generate_study_path=True,
)

# Poll for progress
status = client.document_upload.get_status(result.message_id)
print(f"Status: {status.status}")                     # "queued", "processing", "completed", or "failed"
print(f"Step: {status.step} ({status.step_number}/{status.total_steps})")
print(f"Progress: {status.percent_complete}%")
print(f"Detail: {status.detail}")                      # e.g. "Generated notes for topic 3/7"
print(f"Pipeline: {status.pipeline_type}")             # e.g. "pdf", "youtube", "image"

DocumentUploadStatus Fields

Field	Type	Description
request_id	str	The request ID from send()
status	str	"queued", "processing", "completed", "failed", or "unknown"
step	str	Current step name (e.g. "downloading", "embedding", "generating_notes")
step_number	int	Current step index (1-based)
total_steps	int	Total steps for this pipeline type
percent_complete	int	Overall progress 0-100
pipeline_type	str	Pipeline chosen after routing (e.g. "pdf", "youtube", "image")
detail	str	Sub-step detail (e.g. "Generated notes for topic 3/7")
started_at	str	ISO timestamp when processing started
updated_at	str	ISO timestamp of last status update
error	str	Error message if status is "failed"

Status data is stored in Redis with a 1-hour TTL, reset on every update. If get_status() is called with an unknown request ID or after the TTL expires, it returns status="unknown".

Document Notes (Note Generation)

Enqueue note generation jobs after clustering completes. In the three-queue architecture, Queue 1 (document-upload) handles extraction and clustering, Queue 2 (document-notes) generates notes and persists materials, and Queue 3 (document-persist) links materials to topics after user approval.

Queue 1 automatically enqueues to document-notes after clustering. The notes queue generates notes, persists materials, and sends the callback that closes the upload modal.

Fire and forget

result = client.document_notes.send(
    uploaded_file_id="file_abc123",
    study_set_id="set_xyz",
    user_id="user_123",
    stash_key="clustering-stash/file_abc123/1720000000.json",
    is_deferred=True,
    callback_url="https://api.example.com/completeCluster",
    callback_api_key="secret",
)

print("Enqueued:", result.message_id)

With notes disabled

result = client.document_notes.send(
    uploaded_file_id="file_abc123",
    study_set_id="set_xyz",
    user_id="user_123",
    stash_key="clustering-stash/file_abc123/1720000000.json",
    is_deferred=False,
    generate_notes=False,
    folder_id="folder_abc",
    callback_url="https://api.example.com/completeCluster",
    callback_api_key="secret",
)

Document Notes Fields

Field	Type	Required	Description
uploaded_file_id	str	Yes	ID of the uploaded file record
study_set_id	str	Yes	Study set to persist materials into
user_id	str	Yes	User who uploaded the document
stash_key	str	Yes	GCS key of the clustering stash JSON
is_deferred	bool	No	Whether the study set has existing topics requiring approval (default: False)
generate_notes	bool	No	Whether to generate notes (default: True). Set to False for noNotes materials
callback_url	Optional[str]	No	URL to POST results to after completion
callback_api_key	Optional[str]	No	Auth key for the callback
folder_id	Optional[str]	No	Pre-created folder ID for the materials

Response Types

Method	Return Type	Fields
send()	SendResult	message_id

Document Persist (Topic Linking)

Enqueue topic-linking jobs after user approval. In the three-queue architecture, Queue 3 (document-persist) loads the material map from the stash sidecar and links pre-generated materials to approved topics.

For new study sets (no existing study path), Queue 2 handles everything. For existing study sets, the approve route enqueues to document-persist after the user reviews pending changes.

Fire and forget

result = client.document_persist.send(
    uploaded_file_id="file_abc123",
    study_set_id="set_xyz",
    user_id="user_123",
    stash_key="clustering-stash/file_abc123/1720000000.json",
    callback_url="https://api.example.com/completeCluster",
    callback_api_key="secret",
)

print("Enqueued:", result.message_id)

With approved topic filter

result = client.document_persist.send(
    uploaded_file_id="file_abc123",
    study_set_id="set_xyz",
    user_id="user_123",
    stash_key="clustering-stash/file_abc123/1720000000.json",
    approved_topic_ids=["topic_1", "topic_2", "topic_5"],
    language="en",
    features=["flashcardsSet", "practiceTest"],
    flashcard_amount=12,
    callback_url="https://api.example.com/completeCluster",
    callback_api_key="secret",
)

Document Persist Fields

Field	Type	Required	Description
uploaded_file_id	str	Yes	ID of the uploaded file record
study_set_id	str	Yes	Study set to persist materials into
user_id	str	Yes	User who uploaded the document
stash_key	str	Yes	GCS key of the clustering stash JSON
language	str	No	Target language code (default: "en")
callback_url	Optional[str]	No	URL to POST results to after persistence
callback_api_key	Optional[str]	No	Auth key for the callback
approved_topic_ids	Optional[list[str]]	No	Filter to only these topic IDs from the stash (omit to persist all)
features	Optional[dict]	No	Feature flags for post-processing
flashcard_amount	Optional[int]	No	Number of flashcards to generate
selected_questions	Optional[list]	No	Pre-selected questions for test generation
upload_batch_id	Optional[str]	No	Batch ID for multi-file uploads
splitting	Optional[str]	No	Splitting strategy
typeof_notes	Optional[str]	No	Note generation mode
file_type	Optional[str]	No	MIME type
effective_file_type	Optional[str]	No	Resolved file type
url	Optional[str]	No	Source file URL
filename	Optional[str]	No	Original filename

Response Types

Method	Return Type	Fields
send()	SendResult	message_id

Prompt Ingest (LLM Call Logging)

Send LLM prompt records to the prompt-analyze-api consumer for storage and analysis. This is designed as a fire-and-forget call from within your LiteLLM router (or any LLM orchestration layer) so that logging never blocks inference.

Fire and forget

result = client.prompt_ingest.send(
    call_type="chat",
    request_payload={
        "format": "chat",
        "messages": [
            {"role": "system", "content": "You are a helpful tutor."},
            {"role": "user", "content": "Explain photosynthesis."},
        ],
    },
    status="success",
    model_requested="gpt-4o",
    model_deployed="gpt-4o-2024-08-06",
    provider="openai",
    params={"temperature": 0.7},
    metadata={"user_id": "user_abc", "session_id": "sess_123"},
    tags=["production", "tutor"],
    response_payload={
        "format": "chat",
        "message": {"role": "assistant", "content": "Photosynthesis is..."},
    },
    usage={"input_tokens": 45, "output_tokens": 120, "total_tokens": 165},
    latency_ms=1200,
    cost_usd=0.003,
    content_hash="a1b2c3d4...",
)

print("Enqueued:", result.message_id)

Send and wait for confirmation

result = client.prompt_ingest.send_and_wait(
    call_type="chat",
    request_payload={"format": "chat", "messages": [...]},
    status="success",
    model_requested="gpt-4o",
    timeout=10,
)

print(result.success)
print(result.message_id)

Prompt Ingest Fields

Field	Type	Required	Description
call_type	str	Yes	Type of LLM call: "chat", "text_completion", or "responses"
request_payload	dict	Yes	Structured request (messages, prompt, or input)
status	str	Yes	"success" or "failure"
request_id	Optional[str]	No	Unique call ID (e.g. litellm_call_id)
model_requested	Optional[str]	No	Model name from the request
model_deployed	Optional[str]	No	Actual model/deployment used
provider	Optional[str]	No	Provider name (e.g. "openai", "anthropic")
params	Optional[dict]	No	Model params (temperature, top_p, etc.)
metadata	Optional[dict]	No	Arbitrary metadata (user_id, session, etc.)
tags	Optional[list[str]]	No	Filterable tags
response_payload	Optional[dict]	No	Structured response
response_raw	Optional[dict]	No	Full raw provider response
usage	Optional[dict]	No	Token usage (input_tokens, output_tokens, total_tokens)
error	Optional[str]	No	Error message (when status is "failure")
latency_ms	Optional[int]	No	End-to-end call duration in milliseconds
cost_usd	Optional[float]	No	Estimated cost in USD
content_hash	Optional[str]	No	SHA-256 hash of normalized input text

Response Types

Method	Return Type	Fields
send()	SendResult	message_id
send_and_wait()	QueueResponse	success, message_id, error?, processed_at?

Query (Direct Key Lookups)

Read values directly from Redis by key. These are plain GET / MGET calls, not queue operations.

Get a single key

value = client.query.get("mykey")
print(value)  # str or None

Get multiple keys

values = client.query.get_many(["key1", "key2", "key3"])
print(values)  # list[str | None]

Query Methods

Method	Return Type	Description
get(key)	Optional[str]	Fetch a single key via Redis GET
get_many(keys)	list[Optional[str]]	Fetch multiple keys via Redis MGET

Cleanup

client.disconnect()