sf-queue-sdk 0.2.0

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

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()