Database Operations Skill

Description

Manages database operations across SQLite (local state) and Supabase (remote storage) for the OpenGov Harvester.

Triggers

•"query database"
•"update inventory"
•"check extraction status"
•"sync to supabase"
•"database migration"

Database Architecture

SQLite (Local State)

Path: data/db/opengov_state.db Mode: WAL (Write-Ahead Logging) Purpose: Local state management, queue coordination, session persistence

Key Tables:

•opengov_projects: Project inventory with extraction status
•opengov_opportunities: Opportunity records
•opengov_followers: Follower data
•session_state: Auth session persistence
•extraction_queue: Work queue with priorities

Supabase (Remote Storage)

Project: uzczjsmhitnwtsmmffqc (1st-Commercial-Credit) Region: us-west-2 Schemas: public, opengov

Key Tables:

•public.users: User accounts with RBAC
•public.user_groups: Group assignments
•opengov.api_discovery: Discovered API endpoints
•opengov.project_metadata: Project details with versioning

Common Operations

SQLite Queries

Check Extraction Status:

bash

sqlite3 data/db/opengov_state.db "
  SELECT
    project_id,
    name,
    extracted,
    updated_at
  FROM opengov_projects
  WHERE extracted = 1
  ORDER BY updated_at DESC
  LIMIT 10;
"

Count Opportunities:

bash

sqlite3 data/db/opengov_state.db "
  SELECT COUNT(*) as total_opportunities
  FROM opengov_opportunities;
"

View Queue Status:

bash

sqlite3 data/db/opengov_state.db "
  SELECT
    status,
    COUNT(*) as count
  FROM extraction_queue
  GROUP BY status;
"

Supabase Operations

Pull Remote Schema:

bash

supabase db pull

Create Migration:

bash

supabase migration new add_feature_name

Apply Migrations Locally:

bash

supabase db reset

Push to Remote:

bash

supabase db push

Execute SQL Query:

bash

# Via MCP tool (preferred)
mcp__plugin_supabase_supabase__execute_sql

# Via psql
PGPASSWORD=postgres psql -h 127.0.0.1 -p 54322 -U postgres -d postgres -c "SELECT * FROM opengov.project_metadata LIMIT 5;"

Python Database Access

SQLite Context Manager

python

import sqlite3

def get_extraction_status(project_id: str) -> bool:
    conn = sqlite3.connect('data/db/opengov_state.db')
    conn.execute('PRAGMA journal_mode=WAL')

    try:
        with conn:
            result = conn.execute(
                "SELECT extracted FROM opengov_projects WHERE project_id = ?",
                (project_id,)
            ).fetchone()
            return bool(result[0]) if result else False
    finally:
        conn.close()

Supabase Client

python

from supabase import create_client

url = os.getenv("SUPABASE_URL")
key = os.getenv("SUPABASE_SERVICE_ROLE_KEY")
supabase = create_client(url, key)

# Insert data
data = supabase.table("opengov.project_metadata").insert({
    "project_id": "abc123",
    "name": "City Project",
    "version": 1
}).execute()

# Query with filters
projects = supabase.table("opengov.project_metadata")\
    .select("*")\
    .eq("extracted", True)\
    .limit(10)\
    .execute()

Migration Management

Creating Migrations

Schema Change:

sql

-- supabase/migrations/20260125_add_version_tracking.sql

-- Add version column to track changes
ALTER TABLE opengov.project_metadata
ADD COLUMN version INTEGER DEFAULT 1;

-- Create index for version queries
CREATE INDEX idx_project_version ON opengov.project_metadata(project_id, version);

-- Update RLS policies if needed
ALTER POLICY "Users can read their projects" ON opengov.project_metadata
  USING (user_id = auth.uid() OR is_public = true);

Data Migration:

sql

-- supabase/migrations/20260125_backfill_versions.sql

-- Backfill version numbers for existing records
UPDATE opengov.project_metadata
SET version = 1
WHERE version IS NULL;

Migration Workflow

•✅ Create migration: supabase migration new <name>
•✅ Write SQL in supabase/migrations/<timestamp>_<name>.sql
•✅ Test locally: supabase db reset
•✅ Verify data integrity
•✅ Push to remote: supabase db push

Related Rules

•database-transaction-safety.md: Safe transaction patterns
•etl-atomic-writes.md: Atomic data operations
•data-quality-checks.md: Data validation
•data-lineage-metadata.md: Metadata tracking
•etl-idempotency-patterns.md: Safe re-runs

Inspection & Monitoring

Database Stats

bash

# Supabase stats
supabase inspect db db-stats
supabase inspect db table-stats
supabase inspect db index-stats

# SQLite stats
sqlite3 data/db/opengov_state.db "
  SELECT
    name,
    (SELECT COUNT(*) FROM sqlite_master WHERE type='index' AND tbl_name=m.name) as indexes,
    (SELECT COUNT(*) FROM pragma_table_info(m.name)) as columns
  FROM sqlite_master m
  WHERE type='table';
"

Performance Monitoring

bash

# Long-running queries
supabase inspect db long-running-queries

# Query outliers
supabase inspect db outliers

# Blocking queries
supabase inspect db blocking

Backup & Recovery

SQLite Backup

bash

# Create backup
sqlite3 data/db/opengov_state.db ".backup data/db/backups/opengov_$(date +%Y%m%d_%H%M%S).db"

# Restore from backup
sqlite3 data/db/opengov_state.db ".restore data/db/backups/opengov_20260125_120000.db"

Supabase Backup

bash

# Dump schema
supabase db dump -f supabase/backups/schema_$(date +%Y%m%d).sql

# Dump data
supabase db dump --data-only -f supabase/backups/data_$(date +%Y%m%d).sql

Troubleshooting

Database Locked

Symptom: "database is locked" error Solution:

•Enable WAL mode: PRAGMA journal_mode=WAL
•Close idle connections
•Use context managers for automatic connection cleanup

Migration Conflicts

Symptom: Remote schema diverged from local Solution:

•Pull remote: supabase db pull
•Review conflicts
•Create merge migration
•Test locally: supabase db reset
•Push: supabase db push

Connection Failures

Symptom: Cannot connect to Supabase Solution:

•Check services: supabase status
•Restart: supabase stop && supabase start
•Verify environment variables
•Check network connectivity

Best Practices

Transaction Management

python

# Always use transactions for multi-step operations
with conn:
    conn.execute("INSERT INTO projects ...")
    conn.execute("UPDATE queue SET status ...")
    # Automatic commit on success, rollback on error

Query Optimization

python

# Use prepared statements
cursor.execute(
    "SELECT * FROM projects WHERE id = ?",
    (project_id,)
)

# Batch inserts
cursor.executemany(
    "INSERT INTO opportunities VALUES (?, ?, ?)",
    opportunity_batch
)

Error Handling

python

from sqlite3 import IntegrityError, OperationalError

try:
    with conn:
        conn.execute("INSERT ...")
except IntegrityError:
    logger.warning("Duplicate record, skipping")
except OperationalError:
    logger.error("Database locked, retrying with backoff")
    # Implement exponential backoff