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

Option 1: Using Docker Compose (Recommended)

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

WebSocket Connection Rejected

Check that your authentication token is valid and properly formatted.

// Correct format
const socket = new WebSocket('ws://localhost:8030/ws?token=YOUR_TOKEN');

Database Connection Issues

If you’re seeing database connection errors, ensure Supabase is running:

# Check Supabase status
docker ps | grep supabase

# Restart Supabase if needed
./scripts/supabase-start.sh

Celery Tasks Not Processing

Ensure RabbitMQ and Celery workers are running:

# Check RabbitMQ status
docker ps | grep rabbitmq

# Start RabbitMQ
./scripts/rabbitmq-start.sh

# Start Celery workers
celery -A celery_manager.celery_app worker --loglevel=info

Get Started

Core Concepts

Development

Development Environment Setup

Prerequisites

Local Setup

Clone the Repository

Environment Setup Options

Option 1: Using Docker Compose (Recommended)

Option 2: Local Python Environment

Environment Configuration

Core Development Commands

WebSocket Development

Testing WebSocket Connections

Implementing Custom Message Handlers

AI Integration

Supported AI Providers

Multi-tenancy

Common Issues & Solutions

Additional Resources

​Development Environment Setup

​Prerequisites

​Local Setup

​Clone the Repository

​Environment Setup Options

​Option 1: Using Docker Compose (Recommended)

​Option 2: Local Python Environment

​Environment Configuration

​Core Development Commands

​WebSocket Development

​Testing WebSocket Connections

​Implementing Custom Message Handlers

​AI Integration

​Supported AI Providers

​Multi-tenancy

​Common Issues & Solutions

​Additional Resources

Development Environment Setup

Prerequisites

Local Setup

Clone the Repository

Environment Setup Options

Option 1: Using Docker Compose (Recommended)

Option 2: Local Python Environment

Environment Configuration

Core Development Commands

WebSocket Development

Testing WebSocket Connections

Implementing Custom Message Handlers

AI Integration

Supported AI Providers

Multi-tenancy

Common Issues & Solutions

Additional Resources