Database Operations Skill

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:

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:

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

View Queue Status:

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

Supabase Operations

Pull Remote Schema:

supabase db pull

Create Migration:

supabase migration new add_feature_name

Apply Migrations Locally:

supabase db reset

Push to Remote:

supabase db push

Execute SQL Query:

# 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

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

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:

-- 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:

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

1. ✅ Create migration: supabase migration new <name>
2. ✅ Write SQL in supabase/migrations/<timestamp>_<name>.sql
3. ✅ Test locally: supabase db reset
4. ✅ Verify data integrity
5. ✅ 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

# 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

# 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

# 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

# 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:

1. Enable WAL mode: PRAGMA journal_mode=WAL
2. Close idle connections
3. Use context managers for automatic connection cleanup

Migration Conflicts

Symptom: Remote schema diverged from local Solution:

1. Pull remote: supabase db pull
2. Review conflicts
3. Create merge migration
4. Test locally: supabase db reset
5. Push: supabase db push

Connection Failures

Symptom: Cannot connect to Supabase Solution:

1. Check services: supabase status
2. Restart: supabase stop && supabase start
3. Verify environment variables
4. Check network connectivity

Best Practices

Transaction Management

# 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

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

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

Error Handling

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