Hey there, fellow developer! Ready to supercharge your CRM game with Streak? Let's dive into building a robust Python integration with the Streak API. This powerhouse combo will let you automate workflows, manage pipelines, and handle data like a pro. Buckle up!
Before we jump in, make sure you've got:
requests
libraryFirst things first, let's get our ducks in a row:
pip install requests
Now, let's keep that API key safe:
import os os.environ['STREAK_API_KEY'] = 'your_api_key_here'
Time to test the waters:
import requests api_key = os.environ['STREAK_API_KEY'] base_url = 'https://www.streak.com/api/v1' response = requests.get(f'{base_url}/users/me', auth=(api_key, '')) print(response.json())
If you see your user info, you're golden!
Let's get our hands dirty with some CRUD operations:
def get_pipelines(): response = requests.get(f'{base_url}/pipelines', auth=(api_key, '')) return response.json()
def create_box(pipeline_key, name): data = {'name': name} response = requests.post(f'{base_url}/pipelines/{pipeline_key}/boxes', json=data, auth=(api_key, '')) return response.json()
def update_box_field(box_key, field_key, value): data = {field_key: value} response = requests.post(f'{base_url}/boxes/{box_key}/fields', json=data, auth=(api_key, '')) return response.json()
def delete_box(box_key): response = requests.delete(f'{base_url}/boxes/{box_key}', auth=(api_key, '')) return response.status_code == 200
Ready to level up? Let's tackle some advanced stuff:
def search_boxes(query): params = {'query': query} response = requests.get(f'{base_url}/search', params=params, auth=(api_key, '')) return response.json()
def upload_attachment(box_key, file_path): with open(file_path, 'rb') as file: files = {'file': file} response = requests.post(f'{base_url}/boxes/{box_key}/attachments', files=files, auth=(api_key, '')) return response.json()
from flask import Flask, request app = Flask(__name__) @app.route('/webhook', methods=['POST']) def handle_webhook(): data = request.json # Process webhook data return '', 200 if __name__ == '__main__': app.run(port=5000)
Don't let those pesky errors catch you off guard:
import time def api_request(method, endpoint, **kwargs): while True: try: response = requests.request(method, f'{base_url}/{endpoint}', auth=(api_key, ''), **kwargs) response.raise_for_status() return response.json() except requests.exceptions.HTTPError as e: if e.response.status_code == 429: time.sleep(int(e.response.headers.get('Retry-After', 5))) else: raise
Let's put it all together with a simple lead tracker:
def lead_tracker(): pipeline_key = get_pipelines()[0]['key'] # Assume first pipeline is for leads while True: name = input("Enter lead name (or 'q' to quit): ") if name.lower() == 'q': break box = create_box(pipeline_key, name) update_box_field(box['key'], 'Status', 'New Lead') print(f"Lead '{name}' added successfully!") lead_tracker()
Always test your code! Here's a quick unit test example:
import unittest class TestStreakAPI(unittest.TestCase): def test_get_pipelines(self): pipelines = get_pipelines() self.assertIsInstance(pipelines, list) self.assertTrue(len(pipelines) > 0) if __name__ == '__main__': unittest.main()
And there you have it! You're now armed with the knowledge to build a killer Streak API integration in Python. Remember, this is just the tip of the iceberg. Dive into the Streak API docs for more advanced features and keep experimenting.
Happy coding, and may your pipelines always be full!