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"}],
trigger_automations=True,
)
# 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",
},
},
],
trigger_automations=True,
)
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",
},
},
],
trigger_automations=True,
)
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
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: