​
Development Environment Setup

This guide will help you set up your development environment for working with the ProductFlo API, with a focus on integrating with our WebSocket and AI capabilities.

​
Prerequisites

Before you begin, make sure you have the following installed:

  • Python 3.11 or higher (Python 3.12 recommended)
  • Redis (for WebSocket state and cache management)
  • PostgreSQL 14+ (for database operations)
  • Node.js 18+ (for testing WebSocket clients)
  • Docker and Docker Compose (for containerized development)

​
Local Setup

​
Clone the Repository

git clone https://github.com/productflo/productflo-api.git
cd productflo-api

​
Environment Setup Options

The simplest way to get started is with Docker Compose:

docker-compose up -d

This will start all required services:

  • FastAPI application
  • Redis
  • PostgreSQL (Supabase)
  • RabbitMQ
  • Celery workers
  • Flower monitoring

The API will be available at http://localhost:8030.

​
Option 2: Local Python Environment

# Create a virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies with UV
uv pip install -r requirements.txt

# Start supporting services
./scripts/supabase-start.sh
./scripts/rabbitmq-start.sh

# Start the API server
uvicorn main:app --reload --port 8030

​
Environment Configuration

Copy the example environment file and update it with your settings:

cp .env.example .env

Update the following key variables in the .env file:

ENVIRONMENT=DEV
DEV_SUPABASE_URL=http://localhost:54321
DEV_SUPABASE_KEY=your-supabase-key
DEV_REDIS_URL=redis://localhost:6379/0
DEV_OPENAI_API_KEY=your-openai-api-key
DEV_ANTHROPIC_API_KEY=your-anthropic-api-key

​
Core Development Commands

# Start the API server
uvicorn main:app --reload --port 8030

# Start Celery worker for background tasks
celery -A celery_manager.celery_app worker --loglevel=info

# Run tests
pytest tests/api/ -v

# Run specific test file
pytest tests/api/auth_test.py -v

# Lint code
ruff check .
black .

# Type check
mypy .

​
WebSocket Development

Our API implements both native WebSocket and Socket.IO protocols with a unified interface. The system is designed around room-based communication for efficient message routing.

​
Testing WebSocket Connections

You can use the WebSocket testing endpoint for development:

wscat -c ws://localhost:8030/ws/test

Send a test message:

{
  "type": "chat",
  "room_id": "test-room",
  "data": {
    "message": "Hello world!",
    "sender": "dev-user"
  }
}

​
Implementing Custom Message Handlers

The WebSocket system is implemented with the WebSocketManager class in utils/websocket_manager.py. To implement a custom message handler:

# In utils/websocket_manager.py
async def handle_custom_message(self, user_id: str, message: Dict[str, Any]):
    # Process the message
    response = {
        "type": "custom_response",
        "data": {
            # Your response data
        }
    }
    
    # Send response back to the user
    await self.send_personal_message(user_id, response)

# Register the handler
def _get_message_handler(self, type: str):
    return {
        # Existing handlers...
        "custom_type": self.handle_custom_message,
    }.get(type)

​
AI Integration

The ProductFlo API integrates with multiple LLM providers through a unified interface:

# Example: Using the Agent system
from ml_ai.agent.master.agent import Agent

# Initialize agent with specific configuration
agent = Agent(
    model="smart",  # Options: fast, medium, smart
    provider="anthropic",  # Options: openai, anthropic, groq, deepseek
)

# Generate a response
response = await agent.generate(
    prompt="Describe the key features of a good PCB design",
    context={
        "product_type": "electronic device",
        "industry": "consumer electronics"
    }
)

​
Supported AI Providers

  • OpenAI (GPT-4o, GPT-4, GPT-3.5-Turbo)
  • Anthropic (Claude 3 Opus, Sonnet, Haiku)
  • Groq (Llama-3-70b, Mixtral-8x7b)
  • Deepseek (DeepSeek-Coder)
  • Google (Gemini Pro, Gemini Ultra)

​
Multi-tenancy

The ProductFlo API implements schema-based tenant isolation. Each tenant gets their own PostgreSQL schema within the same database.

# Example: Accessing tenant-specific data
from middleware.tenantHandler import get_tenant_connection

async def get_user_products(tenant_id: str, user_id: str):
    # Get connection for specific tenant
    conn = await get_tenant_connection(tenant_id)
    
    # Query tenant-specific schema
    result = await conn.fetch(
        "SELECT * FROM products WHERE user_id = $1", 
        user_id
    )
    
    return result

​
Common Issues & Solutions

​
Additional Resources