Prerequisites
Before you begin, make sure you have:
- Installed the SDK
- Set up your API key
Initialize the Client
Synchronous Client
Create a new client instance for synchronous operations:
from brew_sdk import BrewSDK
client = BrewSDK()
Async Client
For async applications, use AsyncBrewSDK:
from brew_sdk import AsyncBrewSDK
client = AsyncBrewSDK()
The client automatically uses the BREW_SDK_API_KEY environment variable for authentication.
Import contacts into your Brew audience:
from brew_sdk import BrewSDK
client = BrewSDK()
result = client.contacts.import_.create(
contacts=[
{
"email": "john@example.com",
"first_name": "John",
"last_name": "Doe",
"subscribed": True,
"custom_fields": {
"company": "Acme Inc",
"role": "Developer",
},
},
{
"email": "jane@example.com",
"first_name": "Jane",
"last_name": "Smith",
},
],
)
print("Import results:", result.data.stats)
# Stats(total=2, valid=2, invalid=0)
import asyncio
from brew_sdk import AsyncBrewSDK
async def main():
client = AsyncBrewSDK()
result = await client.contacts.import_.create(
contacts=[
{
"email": "john@example.com",
"first_name": "John",
"last_name": "Doe",
"subscribed": True,
"custom_fields": {
"company": "Acme Inc",
"role": "Developer",
},
},
{
"email": "jane@example.com",
"first_name": "Jane",
"last_name": "Smith",
},
],
)
print("Import results:", result.data.stats)
asyncio.run(main())
Note that the import resource is accessed via import_ (with underscore) because import is a reserved keyword in Python.
Update an existing contact’s information:
updated = client.contacts.update(
email="john@example.com",
first_name="Johnny",
custom_fields={
"role": "Senior Developer",
"department": "Engineering",
},
)
print("Contact updated:", updated.data)
updated = await client.contacts.update(
email="john@example.com",
first_name="Johnny",
custom_fields={
"role": "Senior Developer",
"department": "Engineering",
},
)
print("Contact updated:", updated.data)
Send a Transactional Email
Send a transactional email using a published template:
response = client.send.transactional.send(
chat_id="your-template-id",
to="user@example.com",
variables={
"firstName": "John",
"orderNumber": "ORD-12345",
"trackingUrl": "https://example.com/track/xyz",
},
)
print("Email sent:", response.data)
# Data(sent=1, failed=0, results=[...])
response = await client.send.transactional.send(
chat_id="your-template-id",
to="user@example.com",
variables={
"firstName": "John",
"orderNumber": "ORD-12345",
"trackingUrl": "https://example.com/track/xyz",
},
)
print("Email sent:", response.data)
You can also send to multiple recipients:
response = client.send.transactional.send(
chat_id="your-template-id",
to=["user1@example.com", "user2@example.com"],
variables={
"announcement": "New feature released!",
},
)
Remove a contact from your audience:
deleted = client.contacts.delete(email="john@example.com")
print("Contact deleted:", deleted.data)
# Data(deleted=True, email='john@example.com')
deleted = await client.contacts.delete(email="john@example.com")
print("Contact deleted:", deleted.data)
Working with Types
The SDK uses Pydantic models for type-safe responses:
from brew_sdk import BrewSDK
from brew_sdk.types.contacts import ImportCreateResponse
client = BrewSDK()
result: ImportCreateResponse = client.contacts.import_.create(
contacts=[{"email": "john@example.com", "first_name": "John"}],
)
# Access typed data
print(result.data.stats.total if result.data.stats else None)
print(result.data.import_id)
# Convert to dict
data_dict = result.to_dict()
# Convert to JSON
json_str = result.to_json()
Handling Responses
All SDK methods return Pydantic models with helper methods:
result = client.contacts.import_.create(
contacts=[{"email": "test@example.com"}],
)
# Check success
if result.success:
print("Import ID:", result.data.import_id)
print("Stats:", result.data.stats)
# Handle validation errors
if result.data.errors:
for error in result.data.errors:
print(f"Error at index {error.index}: {error.error}")
# Access metadata
print("Request ID:", result.meta.request_id if result.meta else None)
# Serialize response
print(result.to_json(indent=2))
Error Handling
Wrap API calls in try-except to handle errors:
import brew_sdk
from brew_sdk import BrewSDK
client = BrewSDK()
try:
result = client.contacts.import_.create(
contacts=[{"email": "invalid-email"}],
)
except brew_sdk.BadRequestError as e:
print(f"Invalid request: {e}")
except brew_sdk.AuthenticationError as e:
print("Authentication failed - check your API key")
except brew_sdk.RateLimitError as e:
print("Rate limited - slow down requests")
except brew_sdk.APIStatusError as e:
print(f"API error: {e.status_code} - {e.message}")
See Error Handling for more details.
Complete Example
Here’s a complete example putting it all together:
import brew_sdk
from brew_sdk import BrewSDK
def main():
client = BrewSDK()
try:
# Import a new contact
import_result = client.contacts.import_.create(
contacts=[
{
"email": "newuser@example.com",
"first_name": "New",
"last_name": "User",
"custom_fields": {
"source": "api-quickstart",
},
},
],
)
print("Imported:", import_result.data.stats)
# Send welcome email
email_result = client.send.transactional.send(
chat_id="welcome-email-template",
to="newuser@example.com",
variables={
"firstName": "New",
"dashboardUrl": "https://app.example.com/dashboard",
},
)
print("Welcome email sent:", email_result.data.sent)
except brew_sdk.APIStatusError as e:
print(f"API Error: {e.status_code} - {e.message}")
if __name__ == "__main__":
main()
import asyncio
import brew_sdk
from brew_sdk import AsyncBrewSDK
async def main():
async with AsyncBrewSDK() as client:
try:
# Import a new contact
import_result = await client.contacts.import_.create(
contacts=[
{
"email": "newuser@example.com",
"first_name": "New",
"last_name": "User",
"custom_fields": {
"source": "api-quickstart",
},
},
],
)
print("Imported:", import_result.data.stats)
# Send welcome email
email_result = await client.send.transactional.send(
chat_id="welcome-email-template",
to="newuser@example.com",
variables={
"firstName": "New",
"dashboardUrl": "https://app.example.com/dashboard",
},
)
print("Welcome email sent:", email_result.data.sent)
except brew_sdk.APIStatusError as e:
print(f"API Error: {e.status_code} - {e.message}")
if __name__ == "__main__":
asyncio.run(main())
Next Steps
Contacts
Learn more about managing contacts
Transactional Emails
Send transactional emails
Error Handling
Handle errors and configure retries
API Reference
View the full API documentation
Need Help?
Our team is ready to support you at every step of your journey with Brew. Choose the option that works best for you: