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="user@example.com",
    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="user@example.com",
    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=[
        "student1@example.com",
        "student2@example.com",
        "student3@example.com",
    ],
    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=[
        "student1@example.com",
        "student2@example.com",
        "student3@example.com",
    ],
    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

Optional Fields

# With all optional fields
client.email.send(
    to="support@studyfetch.com",
    preview="Support Request",
    subject="StudyFetch: Support Request",
    paragraphs=["Hi There,", "You received a support request.", issue],
    reply_to="requester@example.com",
    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="user@example.com",
    preview="Welcome!",
    subject="Welcome!",
    paragraphs=["Welcome to the platform!"],
    service="sparkles",
)

# Override the template for a specific email
client.email.send(
    to="user@example.com",
    preview="Welcome!",
    subject="Welcome!",
    paragraphs=["Welcome to the platform!"],
    service="sparkles",
    template="sparkles",
)

Whitelabel Branding

The Honen template accepts optional brand overrides. Empty fields retain the standard Honen identity.

client.email.send(
    to="member@acme.com",
    preview="You have been invited",
    subject="Join Acme",
    paragraphs=["You have been invited to Acme."],
    service="honen",
    template="honen",
    brand={
        "name": "Acme",
        "logoUrl": "https://cdn.acme.com/logo.png",
        "url": "https://learn.acme.com",
        "accentColor": "#0043FF",
    },
)

name also becomes the From display name while the selected route or validated From override continues to control the sending address and provider key.

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

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

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

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

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

Convert Response

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)

Idempotent verified uploads

Pass idempotency_key when retrying a verified upload. Calls with the same environment, queue, and key create one Redis stream entry and return the same message_id.

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,
    idempotency_key="verified-upload:file_abc123",
)

The key is scoped to document-upload and the configured environment. The SDK stores a SHA-256 digest of the caller key in Redis and keeps the idempotency mapping for seven days.

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",
    content_hash="sha256:verified-content-hash",
)

Cancel a document

Record a manual cancellation before or during processing. Cancellation is scoped to the configured environment and retained for seven days.

client.document_upload.cancel_document("file_abc123")

Document Upload Fields

params fields

Response Type

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}")                     # Includes "cancelled" and "superseded" terminal states
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

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/set_xyz/file_abc123.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/set_xyz/file_abc123.json",
    is_deferred=False,
    generate_notes=False,
    folder_id="folder_abc",
    callback_url="https://api.example.com/completeCluster",
    callback_api_key="secret",
)

Note regeneration for existing materials

Use skip_material_creation and existing_material_id to regenerate notes for an existing noNotes material using its original clustering stash. The stash key is deterministic: clustering-stash/{study_set_id}/{uploaded_file_id}.json.

result = client.document_notes.send(
    uploaded_file_id="file_abc123",
    study_set_id="set_xyz",
    user_id="user_123",
    stash_key="clustering-stash/set_xyz/file_abc123.json",
    is_deferred=False,
    generate_notes=True,
    skip_material_creation=True,
    existing_material_id="mat_existing_456",
    callback_url="https://api.example.com/noNotesGen/callback",
    callback_api_key="secret",
)

Document Notes Fields

Response Types

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

Response Types

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

Response Types

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

Cleanup

client.disconnect()