Back

Step by Step Guide to Building a Streak API Integration in Python

Aug 14, 20246 minute read

Introduction

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!

Prerequisites

Before we jump in, make sure you've got:

  • Python 3.7+
  • requests library
  • A Streak API key (grab one from your Streak settings)

Setting up the environment

First 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'

Basic API connection

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!

Core API operations

Let's get our hands dirty with some CRUD operations:

Fetching pipelines

def get_pipelines(): response = requests.get(f'{base_url}/pipelines', auth=(api_key, '')) return response.json()

Creating a box

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

Updating box fields

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

Deleting a box

def delete_box(box_key): response = requests.delete(f'{base_url}/boxes/{box_key}', auth=(api_key, '')) return response.status_code == 200

Advanced features

Ready to level up? Let's tackle some advanced stuff:

Searching for boxes

def search_boxes(query): params = {'query': query} response = requests.get(f'{base_url}/search', params=params, auth=(api_key, '')) return response.json()

Working with attachments

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

Handling webhooks

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)

Error handling and best practices

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

Building a sample application

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

Testing and debugging

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

Conclusion

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!