Integration Mapping and Architecture Visualization

This skill provides expertise in creating visual integration maps and architecture diagrams for Supabase+Airflow+SQLite stacks. It should be used when the user asks about "integration map", "architecture diagram", "wiring diagram", "system visualization", "show connections", or similar visualization topics.

Integration Mapping and Architecture Visualization

Master the creation of visual diagrams showing how Supabase, Airflow, SQLite, and other components are wired together in your system.

Mermaid Diagram Generation

Basic System Architecture Diagram

Generate overview of system components:

def generate_system_architecture_diagram():
    """Generate Mermaid diagram of system architecture."""
    mermaid = [
        "graph TB",
        "    subgraph External",
        "        OpenGov[OpenGov Platform]",
        "    end",
        "",
        "    subgraph Local",
        "        Playwright[Playwright Browser Automation]",
        "        SQLite[(SQLite Local State)]",
        "    end",
        "",
        "    subgraph Supabase",
        "        Postgres[(Postgres Database)]",
        "        Auth[Auth Service]",
        "        Storage[Storage API]",
        "        Functions[Edge Functions]",
        "    end",
        "",
        "    subgraph Orchestration",
        "        Airflow[Apache Airflow]",
        "        Scheduler[Airflow Scheduler]",
        "    end",
        "",
        "    %% Data flow connections",
        "    Playwright -->|Extract Data| SQLite",
        "    SQLite -->|Sync| Postgres",
        "    Airflow -->|Orchestrate| Playwright",
        "    Airflow -->|Store Results| Postgres",
        "    Playwright -->|Authenticate| OpenGov",
        "    Functions -->|Process| Postgres",
        "",
        "    %% Styling",
        "    classDef database fill:#f9f,stroke:#333,stroke-width:2px",
        "    classDef service fill:#bbf,stroke:#333,stroke-width:2px",
        "    classDef orchestration fill:#bfb,stroke:#333,stroke-width:2px",
        "",
        "    class SQLite,Postgres database",
        "    class Playwright,Auth,Storage,Functions service",
        "    class Airflow,Scheduler orchestration",
    ]

    return "\n".join(mermaid)

Data Flow Diagram

Show how data flows through the system:

def generate_data_flow_diagram():
    """Generate Mermaid diagram showing data flow."""
    mermaid = [
        "graph LR",
        "    Start([ETL Pipeline Start]) --> Auth[Authenticate to OpenGov]",
        "    Auth --> Inventory[Load Project Inventory]",
        "    Inventory --> Extract{For Each Project}",
        "    ",
        "    Extract --> ProjectData[Extract Project Details]",
        "    ProjectData --> Opportunities[Extract Opportunities]",
        "    Opportunities --> Followers[Extract Followers]",
        "    ",
        "    ProjectData --> SQLite[(Store in SQLite)]",
        "    Opportunities --> SQLite",
        "    Followers --> SQLite",
        "    ",
        "    SQLite --> Validate[Validate Data]",
        "    Validate --> Transform[Transform for Postgres]",
        "    Transform --> Postgres[(Load to Postgres)]",
        "    ",
        "    Postgres --> Export[Export to CSV/JSON]",
        "    Export --> End([Pipeline Complete])",
        "",
        "    %% Error handling",
        "    Auth -.->|Failed| ErrorLog[Log to run_errors]",
        "    Extract -.->|Failed| ErrorLog",
        "    Validate -.->|Failed| ErrorLog",
    ]

    return "\n".join(mermaid)

Database Schema Relationships

Visualize table relationships:

def generate_schema_diagram(tables):
    """Generate ER diagram from table definitions."""
    mermaid = ["erDiagram"]

    for table_name, table_def in tables.items():
        # Add table with columns
        columns = table_def.get("columns", [])
        for col_name, col_type in columns:
            # Clean column definition
            col_clean = col_name.strip()
            type_clean = col_type.split()[0]  # Get base type
            mermaid.append(f"    {table_name} {{")
            mermaid.append(f"        {type_clean} {col_clean}")
            mermaid.append(f"    }}")

    # Add relationships
    for table_name, table_def in tables.items():
        foreign_keys = table_def.get("foreign_keys", [])
        for fk in foreign_keys:
            # Parse FK: "FOREIGN KEY (project_id) REFERENCES opengov_projects(project_id)"
            if "REFERENCES" in fk:
                parts = fk.split("REFERENCES")
                ref_table = parts[1].strip().split("(")[0].strip()
                mermaid.append(f"    {table_name} ||--o{{ {ref_table} : references")

    return "\n".join(mermaid)

Component Wiring Maps

Airflow to Database Wiring

Show which DAGs connect to which databases:

def generate_airflow_wiring_map(dag_info):
    """Generate diagram showing Airflow DAG database connections."""
    mermaid = [
        "graph LR",
        "    subgraph DAGs",
    ]

    for dag in dag_info:
        mermaid.append(f"        DAG_{dag['id']}[{dag['name']}]")

    mermaid.append("    end")
    mermaid.append("")
    mermaid.append("    subgraph Databases")
    mermaid.append("        SQLite[(SQLite)]")
    mermaid.append("        Postgres[(Postgres)]")
    mermaid.append("    end")
    mermaid.append("")

    # Add connections
    for dag in dag_info:
        if "sqlite" in dag.get("databases", []):
            mermaid.append(f"    DAG_{dag['id']} -->|Read/Write| SQLite")
        if "postgres" in dag.get("databases", []):
            mermaid.append(f"    DAG_{dag['id']} -->|Read/Write| Postgres")

    return "\n".join(mermaid)

Service Dependencies

Show service startup dependencies:

def generate_service_dependency_map():
    """Generate service dependency diagram."""
    mermaid = [
        "graph TD",
        "    Docker[Docker Daemon]",
        "    ",
        "    Docker --> Supabase[Supabase Services]",
        "    Docker --> AirflowDB[(Airflow Postgres)]",
        "    ",
        "    Supabase --> SupabaseDB[(Supabase Postgres)]",
        "    Supabase --> Auth[Auth Service]",
        "    Supabase --> Storage[Storage Service]",
        "    ",
        "    AirflowDB --> Scheduler[Airflow Scheduler]",
        "    AirflowDB --> Webserver[Airflow Webserver]",
        "    ",
        "    Scheduler --> DAGs[DAG Execution]",
        "    DAGs --> SupabaseDB",
        "    DAGs --> SQLite[(SQLite)]",
        "",
        "    %% Styling",
        "    classDef infrastructure fill:#f96,stroke:#333,stroke-width:2px",
        "    classDef database fill:#9cf,stroke:#333,stroke-width:2px",
        "    classDef service fill:#9f9,stroke:#333,stroke-width:2px",
        "",
        "    class Docker infrastructure",
        "    class SupabaseDB,AirflowDB,SQLite database",
        "    class Supabase,Auth,Storage,Scheduler,Webserver,DAGs service",
    ]

    return "\n".join(mermaid)

Auto-Discovery from Codebase

Scan DAG Files for Connections

Automatically detect database connections in DAG files:

def discover_dag_connections():
    """Scan DAG files to discover database connections."""
    dags_folder = os.getenv("AIRFLOW__CORE__DAGS_FOLDER", "dags")
    dag_connections = []

    for dag_file in Path(dags_folder).glob("*.py"):
        connections = analyze_dag_file_connections(dag_file)
        if connections:
            dag_connections.append({
                "file": str(dag_file),
                "dag_id": extract_dag_id(dag_file),
                "databases": connections,
            })

    return dag_connections

def analyze_dag_file_connections(dag_file):
    """Analyze a DAG file to find database connections."""
    connections = set()
    content = dag_file.read_text()

    # Check for SQLite
    if "sqlite3" in content.lower() or "sqlite_db" in content:
        connections.add("sqlite")

    # Check for Postgres/Supabase
    if "postgres" in content.lower() or "supabase" in content.lower():
        connections.add("postgres")

    # Check for specific table references
    if "opengov_projects" in content or "opengov_opportunities" in content:
        connections.add("opengov_schema")

    return list(connections)

Detect API Endpoints

Find all API integrations:

def discover_api_endpoints():
    """Discover external API endpoints from codebase."""
    endpoints = []

    # Search for URL patterns in code
    for python_file in Path("src").rglob("*.py"):
        content = python_file.read_text()

        # Find URLs
        import re
        urls = re.findall(r'https?://[^\s"\']+', content)

        for url in urls:
            if "opengov" in url:
                endpoints.append({
                    "url": url,
                    "file": str(python_file),
                    "type": "opengov_api"
                })
            elif "supabase" in url:
                endpoints.append({
                    "url": url,
                    "file": str(python_file),
                    "type": "supabase_api"
                })

    return endpoints

Interactive Diagrams

Generate D2 Diagrams (Alternative to Mermaid)

D2 provides more sophisticated layouts:

def generate_d2_architecture():
    """Generate D2 diagram syntax."""
    d2 = """
    direction: right

    opengov: OpenGov Platform {
        shape: cloud
        style.fill: "#e1f5ff"
    }

    local: Local Environment {
        playwright: Playwright {
            icon: browser
        }
        sqlite: SQLite {
            shape: cylinder
            icon: database
        }
    }

    supabase: Supabase Cloud {
        postgres: Postgres {
            shape: cylinder
            icon: database
        }
        auth: Auth Service
        storage: Storage API
    }

    airflow: Orchestration {
        scheduler: Airflow Scheduler
        executor: Task Executor
    }

    # Connections
    playwright -> opengov: Extract
    playwright -> sqlite: Store
    sqlite -> postgres: Sync
    airflow -> playwright: Orchestrate
    airflow -> postgres: Results
    """

    return d2

Documentation Generation

Generate Architecture Documentation

Create markdown documentation with embedded diagrams:

def generate_architecture_docs():
    """Generate architecture documentation with diagrams."""
    docs = []

    docs.append("# System Architecture Documentation")
    docs.append("")
    docs.append("## Overview")
    docs.append("")
    docs.append("This document describes the architecture of the OpenGov Harvester system.")
    docs.append("")

    # Add system diagram
    docs.append("## System Components")
    docs.append("")
    docs.append("```mermaid")
    docs.append(generate_system_architecture_diagram())
    docs.append("```")
    docs.append("")

    # Add data flow
    docs.append("## Data Flow")
    docs.append("")
    docs.append("```mermaid")
    docs.append(generate_data_flow_diagram())
    docs.append("```")
    docs.append("")

    # Add component descriptions
    docs.append("## Component Descriptions")
    docs.append("")
    docs.append("### Playwright")
    docs.append("Browser automation for extracting data from OpenGov platform.")
    docs.append("")
    docs.append("### SQLite")
    docs.append("Local database for state management and queueing.")
    docs.append("")
    docs.append("### Supabase")
    docs.append("Remote Postgres database for persistent storage.")
    docs.append("")
    docs.append("### Airflow")
    docs.append("Workflow orchestration for scheduling and monitoring ETL jobs.")
    docs.append("")

    return "\n".join(docs)

Export Formats

Export to Multiple Formats

Support various output formats:

def export_integration_map(format="mermaid"):
    """Export integration map in specified format."""
    if format == "mermaid":
        return generate_system_architecture_diagram()

    elif format == "d2":
        return generate_d2_architecture()

    elif format == "json":
        return json.dumps({
            "components": discover_components(),
            "connections": discover_connections(),
            "apis": discover_api_endpoints(),
        }, indent=2)

    elif format == "markdown":
        return generate_architecture_docs()

    elif format == "svg":
        # Convert Mermaid to SVG (requires mermaid-cli)
        mermaid_code = generate_system_architecture_diagram()
        return convert_mermaid_to_svg(mermaid_code)

    else:
        raise ValueError(f"Unsupported format: {format}")

def convert_mermaid_to_svg(mermaid_code):
    """Convert Mermaid diagram to SVG."""
    import subprocess
    import tempfile

    with tempfile.NamedTemporaryFile(mode='w', suffix='.mmd', delete=False) as f:
        f.write(mermaid_code)
        mmd_file = f.name

    svg_file = mmd_file.replace('.mmd', '.svg')

    subprocess.run([
        'mmdc',  # mermaid-cli
        '-i', mmd_file,
        '-o', svg_file
    ], check=True)

    with open(svg_file, 'r') as f:
        svg_content = f.read()

    os.remove(mmd_file)
    os.remove(svg_file)

    return svg_content

Best Practices

1. Auto-generate when possible - Scan codebase rather than manual documentation
2. Version diagrams - Track diagram changes alongside code changes
3. Keep diagrams focused - Separate overview, data flow, and detailed views
4. Use consistent styling - Apply uniform colors and shapes for component types
5. Update regularly - Regenerate diagrams when architecture changes
6. Include in CI/CD - Validate diagrams build successfully
7. Export to multiple formats - Support different use cases (docs, presentations)

Integration with Garden CLI

Use Garden commands for mapping:

# Generate wiring map
garden system map

# Export to different format
garden system map --format=mermaid > architecture.mmd
garden system map --format=json > integration.json
garden system map --format=markdown > ARCHITECTURE.md

# Generate specific diagram type
garden system map --type=data-flow
garden system map --type=dependencies
garden system map --type=schema

Example Output

Complete Integration Map

graph TB
    subgraph "External Services"
        OpenGov[OpenGov Platform]
    end

    subgraph "Local Development"
        Playwright[Playwright Browser]
        SQLite[(SQLite State DB)]
    end

    subgraph "Supabase Cloud"
        Postgres[(Postgres Database)]
        Auth[Auth Service]
        Storage[Storage API]
        Functions[Edge Functions]
    end

    subgraph "Orchestration"
        Airflow[Airflow Scheduler]
        DAGs[DAG Execution]
    end

    Playwright -->|Extract| OpenGov
    Playwright -->|Store| SQLite
    SQLite -->|Sync| Postgres
    Airflow -->|Schedule| DAGs
    DAGs -->|Execute| Playwright
    DAGs -->|Results| Postgres
    Functions -->|Process| Postgres

    classDef external fill:#ffe6e6
    classDef local fill:#e6f3ff
    classDef cloud fill:#e6ffe6
    classDef orch fill:#fff4e6

    class OpenGov external
    class Playwright,SQLite local
    class Postgres,Auth,Storage,Functions cloud
    class Airflow,DAGs orch